ixChange(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
def user_remain_pay_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)) \
? ? ? ? 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);
? ? print sql_text
# 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)
3、多线程批调度
/Users/nisj/PycharmProjects/EsDataProc/Remain_Data_Proc/BatchThread.py
# -*- coding=utf-8 -*-
import threadpool
from Hive_remain_byWeek_proc import *
# from xx import *
warnings.filterwarnings("ignore")
today = datetime.date.today()
yesterday = today - datetime.timedelta(days=1)
tomorrow = today + datetime.timedelta(days=1)
now_time = time.strftime('%Y-%m-%d %X', time.localtime())
print "当前时间是：",now_time
# 新用户数据先跑出来
last_week = [getNowYearWeek()]
request_newuser_byweek_proc = threadpool.makeRequests(newuser_byweek_proc, last_week)
frist_pool = threadpool.ThreadPool(8)
[frist_pool.putRequest(req) for req in request_newuser_byweek_proc]
frist_pool.wait()
# 然后再执行用户留存和充值金额数据
if True:
? ? batch_week_list = RuningWeekList()
? ? request