正文

用sparkR， 分析上亿条订单数据的脚本。

IP云
IP云V管理员/4 阅读
09/17
提示：扫一扫查出行
【扫一扫了解最新限行尾号】
复制提示

上周我们这个10人的小团队开发的推荐拉新系统， 日拉新人数已接近4万人。过去几个月这个系统从无到有， 拉新从日增几千稳步增长到日增几万， 同事们几个月来，每天工作13个小时以上，洗澡时间都没有， 有时就住在公司， 回家怕吵到家人，只能睡客厅地板， 周日也不能保证休息。 大家的全力投入，不懈努力才能有这个结果。 非常感慨团队产生的的化学反应， 和惊人的生产效率。 产品稳定后，最近全面转入大数据分析， 和机器学习阶段， 开始做真正的增长黑客实践。 spark， R， scala都是刚刚开始深入地学习，没几天， 还好有数据， 学的快！， 不休息， 连做梦都是在做分析数据的工作， 日进千里啊。

刚开始用spark-sql的时候， 如果做一个复杂的查询，写一长串sql， 谁都看不懂，拆成小sql， 就要保存中间结果， 效率低下。 用了几天后， 开始切入sparkR和Scala , 发现效率比直接用spark－sql高太多了， 代码可读性也强太多。此外善用cahe，也可以有效提高效率。

下面都是干货。废话不多少， 只希望帮到你。

工作目标： 分析一下新手券分享的拉新效果和人数，需要对最近15日的订单大概2亿多条订单纪录， 以及300万左右的领券纪录， 几十万笔的返利信息做全库查询 ， 这在msql上是不可能完成的任务。 对spark＋hive来说， 也很耗时， 但一个小时内可以搞定。

用R写了一下查询脚本， 稍后准备改成scala的。 两者都是调用spark api， 区别应该只在语法上。

用15个节点的spark跑这个查询脚本， 大概需要半个多小时才能出来结果。代码是最完整，最准确的文档， 提纲挈领的总结以后得空再总结。

############################statistics.R################################

#领券日期参数, 修改统计日参数
date_parameter <- "2016-07-11"
dayCount_parameter = 1

hiveContext <- sparkRHive.init(sc)
sql(hiveContext, "use honeycomb_bh_db")

#通过hiveSql 获得想要的并集集合并且缓存下来 sql date_add
##程序执行阶段1: 数据准备。。。。。
acquired_users_sql <-"select * from sc_t_acquire_record where sc_t_acquire_record.year=2016 and sc_t_acquire_record.month=07 and to_date(ct_time)='STARTDATE'"
all_order_sql <- "select * from sc_t_order_all_info As a where a.year=2016 and a.month=07 and to_date(a.create_time)>='STARTDATE' and to_date(a.create_time)<=date_add(date('STARTDATE'),14) and product_id=210"
rebate_order_sql <- "select * from sc_t_order_rebate_info As a where a.year=2016 and a.month=07 and to_date(a.create_time)>='STARTDATE' and to_date(a.create_time)<=date_add(date('STARTDATE'),7) and product_id=210"

acquired_users_sql<-sub(pattern='STARTDATE', replacement=date_parameter, acquired_users_sql)
all_order_sql<-gsub(pattern='STARTDATE', replacement=date_parameter, all_order_sql)
rebate_order_sql<-gsub(pattern='STARTDATE', replacement=date_parameter, rebate_order_sql)

#当天领券绑定的用户集合
acquired_users <-sql(hiveContext,acquired_users_sql)
cache(acquired_users)

#15日内的全订单集合
all_orders <-sql(hiveContext,all_order_sql)

#7日内返利的订单集合
rebated_orders <- sql(hiveContext,rebate_order_sql)

#第0日领券后到14日结束前， 有打车纪录的
acquired_users_with_orders<-join(acquired_users,all_orders, acquired_users$presentee_mobile==all_orders$passenger_phone, "left_outer")
acquired_users_with_orders <- filter(acquired_users_with_orders, "passenger_phone is not null")

mobiles_acquired_users <-distinct(select(acquired_users_with_orders, "presentee_mobile"))
#write.json(acquired_users_with_orders, "file:///home/rd/spark/bin/20160711_users_convertion.json")

#第0日领券后～第7日结束前，被返利的领券用户
orders_rebated_within_8days <- join(acquired_users,rebated_orders, acquired_users$presentee_mobile==rebated_orders$passenger_phone, "left_outer")
orders_rebated_within_8days <- filter(orders_rebated_within_8days, "passenger_phone is not null")

cache(orders_rebated_within_8days)
results <- data.frame("name" = c("frist"), "value" = c(0),stringsAsFactors=FALSE)

##程序执行阶段2: 开始利用spark进行集合运算。。。。。

#第0日到第7日结束前， 券有效期内打过车的领券用户订单数据
rules<- "to_date(a.create_time)>='STARTDATE' and to_date(a.create_time)<=date_add(date('STARTDATE'),7)"
rules<-gsub(pattern='STARTDATE', replacement=date_parameter, rules)
orders_within_8days = filter(acquired_users_with_orders, rules)
mobiles_with_orders_within_8days <- distinct(select(orders_within_8days, "presentee_mobile"))

#第8日到第14日结束前， 券过期后， 打过车的领券用户订单数据
rules<- "to_date(a.create_time)>=date_add(date('STARTDATE'),8) and to_date(a.create_time)<=date_add(date('STARTDATE'),15)"
rules<-gsub(pattern='STARTDATE', replacement=date_parameter, rules)
orders_after_8days = filter(acquired_users_with_orders, rules)
mobiles_with_orders_after_8days <- distinct(select(orders_after_8days, "presentee_mobile"))

#第0日到第7日结束前， 被返利信息纪录的领券用户
mobiles_user_reabted <-distinct(select(orders_rebated_within_8days, "presentee_mobile"))

#券0～7天有效期内首单后未被返利的用户
mobiles_my_team_losted <- except(mobiles_with_orders_within_8days, mobiles_user_reabted)

#第8日券有效期过后， 14日内， 有成交纪录被sic统计方法， 统计进来的用户
mobiles_after_7days_countedBySicheng <-except(mobiles_with_orders_after_8days, mobiles_user_reabted)

#券0～7天有效期内首单后未被返利的用户, 第8日到第14日成单， 被sic统计转化的用户
mobiles_my_team_losted_countedBySicheng <-intersect(mobiles_my_team_losted, mobiles_with_orders_after_8days)

#第8日券有效期过后， 14日内， 思成没有统计的首单用户
mobiles_both_losted <- except(mobiles_my_team_losted, mobiles_after_7days_countedBySicheng)

#券0～7天有效期内首单后未被返利， 后7天没打车的用户
mobile_first_order_withno_coupon_no_futher_order_after_7days <- except(mobiles_my_team_losted, mobiles_with_orders_after_8days)

#7日内没打车， 后7日打车的用户
mobiles_with_order_invoked_coupon <- except(mobiles_with_orders_after_8days, mobiles_with_orders_within_8days)

#领券后15天里打车的用户, 由于业务特性，可以重复领券 这个存在重复统计。
mobiles_converted = acquired_users_with_orders

#程序运行阶段： 输出结果。。。
results<-rbind(results, c("领新手券的用户数量", nrow(distinct(select(acquired_users, "presentee_mobile")))))
results<-rbind(results, c("领新手券后15日转化的用户数量", nrow(mobiles_acquired_users)))
results<-rbind(results, c("领新手券7日内打车用券转化的用户数量", nrow(mobiles_user_reabted)))
results<-rbind(results, c("新手券有效期过期后7日内打车转化用户", nrow(mobiles_after_7days_countedBySicheng)))
results<-rbind(results, c("sic统计方法统计的转化用户数", nrow(mobiles_user_reabted)+nrow(mobiles_after_7days_countedBySicheng)))
results<-rbind(results, c("7日内首单未用新手券的人数", nrow(mobiles_my_team_losted)))
results<-rbind(results, c("7日内首单未用新手券, 后7日内没打车的人数", nrow(mobiles_both_losted)))
results<-rbind(results, c("7日内首单未用新手券, 后7日内有打车的人数", nrow(mobiles_my_team_losted_countedBySicheng)))

results<-rbind(results, c("领新手券后7日内未打车, 后7日又打车的人数", nrow(mobiles_with_order_invoked_coupon)))
results