python中调取hive中信息实例

#encoding=utf8
import sys
import datetime
import MySQLdb

sys.path.append("/home/hadoop/hivelib")
from hiveclient import *
from common import *

start_date = (datetime.datetime.today() - datetime.timedelta(1)).strftime('%Y-%m')+'-01' #当月1日
end_date = (datetime.datetime.today() - datetime.timedelta(1)).strftime('%Y-%m-%d') #当月昨天

client = HiveClient()
client.open()

jipiaocps_sql = '''
select le.ptdate,le.fromid,tmp1.orderno,tmp1.wzf,tmp1.yzf,tmp1.ytp
from labrador.leadslog le join (
select k.orderno,k.leadsid,sum(if(k.orderstate <30,1,0)) wzf,
sum(if(k.orderstate>=30 and k.orderstate<=49 and (p.state is NULL or p.state != 22),1,0)) yzf,
sum(if(k.orderstate>=30 and p.state=22,1,0)) ytp
from jipiao.ktepuser k join jipiao.passenger p
on (k.orderno = p.orderno)
group by k.orderno,k.leadsid) tmp1
on (parse_url(le.url,'QUERY','leads_id') = tmp1.leadsid )
where le.ptdate>='%s' and le.ptdate <='%s' and le.ptchannel='jipiao'
and external_source='marketing'
and (tmp1.wzf+tmp1.yzf+tmp1.ytp)>0 ''' %(start_date,end_date)
res_jipiaocps = client.execute(jipiaocps_sql)#查询订单状态

#print jipiaocps_sql
#res_jipiaocps = ["2012-01-01\tKaaaaa-Saaaaa-Taaaaa\t20120101201230101202\t3\t1\t2",]#查询订单状态

DB = {'DB_HOST':'192.168.1.64','DB_USER':'demo','DB_PASSWD':'demo@xwork','DB_PORT':'3306','DB_NAME':'uniondb'} #线上连接
#DB = {'DB_HOST':'192.168.0.100','DB_USER':'kuxun_test','DB_PASSWD':'kuxun_test!#%@$^','DB_PORT':'3306','DB_NAME':'uniondb'} #测试连接
conn = MySQLdb.connect(DB["DB_HOST"], DB["DB_USER"], DB["DB_PASSWD"], DB["DB_NAME"], int(DB["DB_PORT"]), 'utf-8')
cur = conn.cursor()

DBLS = {'DB_HOST':'192.168.2.241','DB_USER':'kteporder_r','DB_PASSWD':'kteporder@op(j3j_RD','DB_PORT':'3306','DB_NAME':'kteporder'}#线上连接
connls = MySQLdb.connect(DBLS["DB_HOST"], DBLS["DB_USER"], DBLS["DB_PASSWD"], DBLS["DB_NAME"], int(DBLS["DB_PORT"]), 'utf-8')
curls = connls.cursor()

delsql = "delete from `jipiaocps` where `cps_date`>='%s' and `cps_date`<='%s'" %(start_date,end_date)
cur.execute(delsql)

for obj in res_jipiaocps:
row_str = obj.split('\t')
cps_date = row_str[0]
trackcode = row_str[1]
order_sn = row_str[2]
wzf_ts = int(row_str[3])
yzf_ts = int(row_str[4])
ytp_ts = int(row_str[5])
if wzf_ts>0:
insql_wzf = "insert into `jipiaocps` (`cps_date`, `trackcode`, `order_sn`, `tickets`, `state`, `acode`) VALUES ('%s', '%s', '%s', %d, 1, '%s')" %(cps_date,trackcode,order_sn,wzf_ts,trackcode.split("-S")[0])
cur.execute(insql_wzf)
if yzf_ts>0:
insql_yzf = "insert into `jipiaocps` (`cps_date`, `trackcode`, `order_sn`, `tickets`, `state`, `acode`) VALUES ('%s', '%s', '%s', %d, 2, '%s')" %(cps_date,trackcode,order_sn,yzf_ts,trackcode.split("-S")[0])
cur.execute(insql_yzf)
if ytp_ts>0:
insql_ytp = "insert into `jipiaocps` (`cps_date`, `trackcode`, `order_sn`, `tickets`, `state`, `acode`) VALUES ('%s', '%s', '%s', %d, 3, '%s')" %(cps_date,trackcode,order_sn,ytp_ts,trackcode.split("-S")[0])
cur.execute(insql_ytp)
if trackcode == "Klashou-S1512341-T1317691":
del_lashou = '''
delete from `jipiaocps` where trackcode='Klashou-S1512341-T1317691' and cps_date='%s' and state=2
''' %(cps_date)
cur.execute(del_lashou)

sql_lashou = '''
select order_sn,passenger_number from `order` where source ='ktep_api_lashou' and order_create_time>='%s 00:00:00' and order_create_time<='%s 23:59:59' and order_state>=30
''' %(cps_date,cps_date)
curls.execute(sql_lashou)
res_lashou = curls.fetchall()
for item in res_lashou:
ins_lashou = '''
insert into `jipiaocps` (`cps_date`, `trackcode`, `order_sn`, `tickets`, `state`, `acode`) VALUES
('%s', 'Klashou-S1512341-T1317691', '%s', %d, 2, 'Klashou')
''' %(cps_date,item[0],item[1])
cur.execute(ins_lashou)
client.close()

#移动出票
for mb in range(1,33):
cps_date = (datetime.datetime.today() - datetime.timedelta(mb)).strftime('%Y-%m-%d') #当月昨天
del_yidong = '''
delete from `jipiaocps` where (trackcode like 'Kqqliulanqi%%' or trackcode like 'Kzhuomian%%') and cps_date='%s'
''' %(cps_date)
cur.execute(del_yidong)

sql_yidong = '''
select source,order_sn,passenger_number from `order` where source like 'kxmb_k%%-s%%-t%%' and order_create_time>='%s 00:00:00' and order_create_time<='%s 23:59:59' and order_state>=30;
''' %(cps_date,cps_date)
curls.execute(sql_yidong)
res_yidong = curls.fetchall()
for item in res_yidong:
tmp = item[0]
for i, j in {'kxmb_k':'K', '-s' : '-S', '-t' : '-T','kxmb_K':'K'}.iteritems():
tcode_str = tmp.replace(i, j)
tmp = tcode_str
acode_str = tcode_str.split('-S')
ins_yidong = '''
insert into `jipiaocps` (`cps_date`, `trackcode`, `order_sn`, `tickets`, `state`, `acode`) VALUES
('%s', '%s', '%s', %d, 2, '%s')
''' %(cps_date,tcode_str,item[1],item[2],acode_str[0])
if acode_str[0]=="Kqqliulanqi":
cur.execute(ins_yidong)
if acode_str[0]=="Kzhuomian":
cur.execute(ins_yidong)