利用HQL统计: 新增用户数，日活，留存率

问题导读

1.每天新增用户数怎么实现？
2.日活怎么实现？
3.留存率怎么实现？







用户行为触发的日志上报,已经存放在Hive的外部分区表中.
结构如下:

new1

主要字段内容：
dt表示日期,如20160510
platform表示平台,只有两个选项,苹果和安卓
mid是用户机器码,类似于网卡MAC地址什么的
pver是版本
channel是分发渠道

现在需要统计每天用户的新增,日活和留存率.
其中
留存率的概念是,如果用户在5月1日第一次使用我们的产品。
如果5月2日他还使用了,那么5月1日的“一日留存”加一.
同理5月3日他又使用了，5月1日的“两日留存”加一.
5月1日的“一日留存率”=5月1日“一日留存” / 5月1日新增用户数量.

先创建一个表,记录用户首次使用的日期.

dt是用户首次使用的日期,比如 20160510
cver是版本
pcid是用户机器码,就是原始日志表的mid

然后创建一个每天数据的存放表,统计昨天一天的新增,日活和留存.

new3

dt是日期
type 1:新增 2:留存 3:日活
num 是用户数量,
dtdiff仅仅用于计算留存,说明用户使用和首次使用的日期间隔多少天.

1.Hive统计每天新增用户

$dt是shell传入的变量
dt=$(date -d last-day +%Y%m%d)
该脚本每天凌晨执行,统计昨天的数据.
每次执行,先清空report_userinfo表
[mw_shl_code=sql,true]insert into user_login_history   
select platform,min(dt),channel,cver,mid,1 from log_vvim  
where   
mid not in (select pcid from user_login_history where type=1)  
and mid is not null  
and dt=$dt  
group by platform,channel,cver,mid;[/mw_shl_code]

这个意思就是 原来没有记录在user_grouproom_login_history表中的pcid,如果出现在昨天的日志表中,则说明用户是新增的.

然后将昨天新增的用户数量写入
[mw_shl_code=sql,true]insert into report_userinfo   
select platform,dt,channel,cver,type,count(*) num,-1 from user_login_history   
where type=1   
and dt=$dt  
group by platform,dt,channel,cver,type;[/mw_shl_code]

2.统计每天活跃用户数量

[mw_shl_code=sql,true]insert into report_userinfo   
select platform,dt,channel,cver,3,count(distinct mid),-1 from log_vvim   
where   
mid is not null and dt=$dt  
group by dt,platform,channel,cver;  [/mw_shl_code]

这个倒是简单,根据原始的日志表,统计今天使用过的pcid,经过去重的用户就是今天的日活用户量.

3.统计留存率.

[mw_shl_code=sql,true]insert into report_userinfo   
select   
xinzeng.platform,  
xinzeng.dt,  
xinzeng.channel,  
xinzeng.cver,  
2,  
count(distinct cunliu.pcid),  
datediff(  
    from_unixtime(unix_timestamp(cast(cunliu.dt as string),'yyyyMMdd')),  
    from_unixtime(unix_timestamp(cast(xinzeng.dt as string),'yyyyMMdd'))  
)  
from  
(  
    select * from user_login_history where type=1  
) xinzeng  
inner join   
(  
    select   
    platform,  
    dt,  
    channel,  
    cver,  
    mid pcid   
    from log_vvim   
    where mid is not null and dt=$dt group by platform,dt,channel,cver,mid  
) cunliu on  
(  
    xinzeng.platform=cunliu.platform and  
    xinzeng.channel=cunliu.channel and  
    xinzeng.cver=cunliu.cver and  
    xinzeng.pcid=cunliu.pcid  
)  
where cunliu.dt>xinzeng.dt   
group by   
xinzeng.platform,xinzeng.dt,xinzeng.channel,xinzeng.cver,  
datediff(  
    from_unixtime(unix_timestamp(cast(cunliu.dt as string),'yyyyMMdd')),  
    from_unixtime(unix_timestamp(cast(xinzeng.dt as string),'yyyyMMdd'))  
);  [/mw_shl_code]
该SQL主要计算昨天使用过的用户,他的首次使用日期,然后计算差值

new4

表示安卓平台,20160425那天首次使用的用户,在8天之后,还使用过的用户数量为20人。

因为计算新增和日活在计算留存之前,
cunliu.dt>xinzeng.dt
主要是确定当天新增的用户不计入留存率计算.

统计完成之后,将hive表导入MySQL
[mw_shl_code=shell,true]sqoop export --connect jdbc:mysql://IP:端口/report --username uname --password "pwd" --table report_userinfo --export-dir '/user/hive/warehouse/logs.db/report_userinfo' --fields-terminated-by '\001'[/mw_shl_code]
最终通过报表展现

您好，咨询下，这种t+1的模式，是将用户访问日志每天一次导入hdfs，如果同时需要实时分析用户访问网站情况，该如何设计离线和实时的日志分析架构咧？

醉半城 发表于 2017-3-10 09:46
您好，咨询下，这种t+1的模式，是将用户访问日志每天一次导入hdfs，如果同时需要实时分析用户访问网站情况 ...

可以先将日志打入到kafka，然后将从kafka中读出的日志进行清洗，清洗完建新的topic再打回kafka，离线部分：kafka+flume-ng+hdfs  ,实时部分：kafka + spark straming / storm 。

感谢分享

feilong 发表于 2017-3-10 09:51
可以先将日志打入到kafka，然后将从kafka中读出的日志进行清洗，清洗完建新的topic再打回kafka，离线部分 ...

谢谢，一般情况是，原始应用的访问日志是写文件的形式，每天一个文件夹。如果改为打入kafka，系统改变会比较大。
如果使用flume的exec source去收集日志，离线：flume -> hdfs ，实时：flume -> kafka -> storm/spark streaming 这种方案怎样？比较担心flume exec source的性能和丢数据情况。

醉半城 发表于 2017-3-10 10:09
谢谢，一般情况是，原始应用的访问日志是写文件的形式，每天一个文件夹。如果改为打入kafka，系统改变会 ...

你给的方案可能会出现 离线和实时数据不一致的情况，尤其是 某一侧流程突然down掉的时候。flume打入到kafka，再在kafka后接离线和实时，就不会出现那种问题。改动的话，flume侧就是改下sink的配置，实时部分属于新增，谈不上对原先流程的影响，离线部分是比原先多了2步，保证这2步没问题就可以了。 flume打入到kafka照样可以用spooldir的形式，另flume也可以从kafka导出数据到hdfs。

feilong 发表于 2017-3-10 10:36
你给的方案可能会出现 离线和实时数据不一致的情况，尤其是 某一侧流程突然down掉的时候。flume打入到kaf ...

嗯，那就是离线：flume -> kafka -> flume -> hdfs  实时：flume -> kafka -> storm/spark streaming ?

醉半城 发表于 2017-3-10 10:55
嗯，那就是离线：flume -> kafka -> flume -> hdfs  实时：flume -> kafka -> storm/spark streaming ?
...

对的。

感谢分享，非常棒

非常棒，只是讨论的有点不明白，为什么flume -> kafka -> flume -> hdfs，kafka为什么还要到flume呢？kafka之后一般是mr或者spark处理数据啊？