newuser_remain_pay_data = os.popen("""source /etc/profile; \
/usr/lib/hive-current/bin/hive -e " \
add jar /home/hadoop/nisj/udf-jar/hadoop_udf_radixChange.jar; \
create temporary function RadixChange as 'com.kascend.hadoop.RadixChange'; \
with his_new_user as (select appsource,appkey,identifier,RadixChange(uid,16,10) uid \
from bi_newuser_byweek \
where pt_week = '%s' \
), \
curr_week_data as (select appsource,appkey,identifier,RadixChange(uid,16,10) uid \
from bi_all_access_log \
where case when weekofyear(pt_day)>=52 and month(pt_day)=1 then concat(year(pt_day)-1,'#',weekofyear(pt_day)) else concat(year(pt_day),'#',weekofyear(pt_day)) end = '%s' \
group by appsource,appkey,identifier,RadixChange(uid,16,10)) \
select a1.appkey,a1.appsource,count(distinct a2.identifier) remain_cnt,0 pay_amount \
from his_new_user a1 \
inner join curr_week_data a2 on a1.appkey=a2.appkey and a1.identifier=a2.identifier and a1.appsource=a2.appsource \
group by a1.appkey,a1.appsource \
;" \
""" % (his_week, curr_week)).readlines();
nrpd_list = []
for nrp_list in newuser_remain_pay_data:
nrp = re.split('\t', nrp_list.replace('\n', ''))
nrpd_list.append(nrp)
for nrpd in nrpd_list:
remain_week = curr_week
appkey = nrpd[0]
appsource = nrpd[1]
remain_cnt = nrpd[2]
pay_amount = nrpd[3]
etl_time = time.strftime('%Y-%m-%d %X', time.localtime())
os.system("""/usr/bin/mysql -hMysqlHost -P6603 -uhadoop -pMysqlPass -e "use funnyai_data; \
insert into bi_user_remain_pay_byweek(data_week,appsource,appkey,remain_week,remain_cnt,pay_amount,etl_time) \
select '%s','%s','%s','%s','%s','%s','%s'; \
" """ % (his_week, appsource, appkey, remain_week, remain_cnt, pay_amount, etl_time))
# Batch Test
# curr_week = '2016#6'
# his_week = '2015#46'
# user_remain_payamount_byweek(curr_week, his_week)
# user_remain_pay_byweek(curr_week, his_week)
# batch_week = '2015#46'
# newuser_byweek_proc(batch_week)
另：供打印sql测试的代码
/Users/nisj/PycharmProjects/EsDataProc/Remain_Data_Proc/xx.py
# -*- coding=utf-8 -*-
import time
import os
import re
from WeekCalc import *
warnings.filterwarnings("ignore")
def newuser_byweek_proc(batch_week):
week1day = getWeekFristday(batch_week)
sql_text = """/usr/lib/hive-current/bin/hive -e " \
alter table bi_newuser_byweek drop if exists partition(pt_week='%s'); \
alter table bi_newuser_byweek add partition(pt_week='%s'); \
insert into table bi_newuser_byweek partition (pt_week='%s') \
select a1.appsource,a1.appkey,a1.identifier,a1.uid from ( \
select appsource,appkey,identifier,uid \
from bi_all_access_log \
where case when weekofyear(pt_day)>=52 and month(pt_day)=1 then concat(year(pt_day)-1,'#',weekofyear(pt_day)) else concat(year(pt_day),'#',weekofyear(pt_day)) end = '%s' \
group by appsource,appkey,identifier,uid) a1 \
left join \
(select appsource,appkey,identifier,uid \
from bi_all_access_log \
where pt_day < '%s' ) a2 on a1.appkey=a2.appkey and a1.identifier=a2.identifier and a1.appsource=a2.appsource and a1.uid=a2.uid \
where a2.identifier is null \
;" \
""" % (batch_week, batch_week, batch_week, batch_week, week1day);
print sql_text
def user_remain_payamount_byweek(curr_week, his_week):
sql_text="""source /etc/profile; \
/usr/lib/hive-current/bin/hive -e " \
add jar /home/hadoop/nisj/udf-jar/hadoop_udf_radixChange.jar; \
create temporary function RadixChange as 'com.kascend.hadoop.RadixChange'; \
with his_new_user as (select appsource,appkey,identifier,RadixChange(uid,16,10) uid \
from bi_newuser_byweek \
where pt_week = '%s' \
), \
curr_week_data as (select appsource,appkey,identifier,RadixChange(uid,16,10) uid \
from bi_all_access_log \
where case when weekofyear(pt_day)>=52 and month(pt_day)=1 then concat(year(pt_day)-1,'#',weekofyear(pt_day)) else concat(year(pt_day),'#',weekofyear(pt_day)) end = '%s' \
group by appsource,appkey,identifier,RadixChange(uid,16,10)), \
curr_week_pay as (select uid,sum(amount) amount \
from data_chushou_pay_info \
where state=0 and \
case when weekofyear(pt_day)>=52 and month(pt_day)=1 then concat(year(pt_day)-1,'#',weekofyear(pt_day)) else concat(year(pt_day),'#',weekofyear(pt_day)) end = '%s' \
group by uid) \
select b1.appkey,b1.appsource,sum(b2.amount) pay_amount from \
(select a1.appkey,a1.appsource,a1.uid \
from his_new_user a1 \
inner join curr_week_data a2 on a1.appkey=a2.appkey and a1.identifier=a2.identifier and a1.appsource=a2.appsource \
group by a1.appkey,a1.appsource,a1.uid) b1 \
left join curr_week_pay b2 on b1.uid=b2.uid \
group by b1.appkey,b1.appsource \
;" \
""" % (his_week, curr_week, curr_week);
print sql_text