留存用户：在前面的blog的指标已经说明，前一个周期新增的用户在当前周期的仍在活跃成为留存用户。

本文采用了三种实现：一种是用.net+sql来实现数据的处理，一种是利用hadoop平台进行数据处理，一种是基于hive的实现

.net + sql

                    #region 构造统计日期临时表DataTable originalDates = new DataTable();DataColumn col = new DataColumn("OriginalStartDate", typeof(Int32));originalDates.Columns.Add(col);col = new DataColumn("OriginalEndDate", typeof(Int32));originalDates.Columns.Add(col);int step = endDate.Subtract(startDate).Days + 1;for (int i = 1; i <= 30; i++){//如果是天的周期，增加到30日留存if (period != PeriodOptions.Daily && i > 6) break;DataRow newRow = originalDates.NewRow();DateTime pStartDate;DateTime pEndDate = Utility.GetNextStatDate(period, startDate, endDate, -i, out pStartDate);newRow["OriginalStartDate"] = int.Parse(pStartDate.ToString("yyyyMMdd"));newRow["OriginalEndDate"] = int.Parse(pEndDate.ToString("yyyyMMdd"));originalDates.Rows.Add(newRow);}#endregion//计算汇总SqlParameter[] paramters = new SqlParameter[]{SqlParamHelper.MakeInParam("@dt", SqlDbType.Structured),SqlParamHelper.MakeInParam("@StartDate", SqlDbType.Int, 4, startDate.ToString("yyyyMMdd")),SqlParamHelper.MakeInParam("@EndDate", SqlDbType.Int, 4, endDate.ToString("yyyyMMdd")),SqlParamHelper.MakeInParam("@Period", SqlDbType.TinyInt, 1, (int)period)                        };paramters[0].TypeName = "dbo.OriginalDatesType";paramters[0].Value = originalDates;DataSet ds = SqlHelper.ExecuteDataset(ComputingDB_ConnString, CommandType.StoredProcedure, "PR_StatRetainedUsers", paramters);

ALTER PROCEDURE [dbo].[PR_StatRetainedUsers] (@dt OriginalDatesType readonly,   @StartDate int,@EndDate int, @Period tinyint)
AS
begincreate table #RetainedUsers(SoftID int,Platform tinyint,ChannelID int,OriginalStatDate int,RetainedUserCount int)declare @sql nvarchar(max);if (@Period = 1) beginset @sql = N'insert into #RetainedUsers(SoftID,Platform,ChannelID,OriginalStatDate,RetainedUserCount)select A.SoftID,A.PLATFORM,A.FirstChannelID,B.OriginalEndDate,COUNT(distinct A.IMEI)from ...' + CAST((@StartDate / 10000) as nvarchar(10)) + N' A with(nolock) inner join @dt Bon A.LoginDate=@StartDate and A.FirstLoginDate between B.OriginalStartDate and B.OriginalEndDate and (FromCache=0 or FromCache is null)group by A.SoftID,A.PLATFORM,A.FirstChannelID,B.OriginalEndDate;'exec sp_executesql @sql, N'@StartDate int,@dt OriginalDatesType readonly', @StartDate, @dt    end else beginif (@StartDate/10000 = @EndDate / 10000) beginset @sql = N'insert into #RetainedUsers(SoftID,Platform,ChannelID,OriginalStatDate,RetainedUserCount)select A.SoftID,A.PLATFORM,A.FirstChannelID,B.OriginalEndDate,COUNT(distinct A.IMEI)from ....' + CAST((@StartDate / 10000) as nvarchar(10)) + N' A with(nolock) inner join @dt Bon A.Part=@part and A.LoginDate between @StartDate and @EndDate and A.FirstLoginDate between B.OriginalStartDate and B.OriginalEndDate and (FromCache=0 or FromCache is null)group by A.SoftID,A.PLATFORM,A.FirstChannelID,B.OriginalEndDate;'end else beginset @sql = N'insert into #RetainedUsers(SoftID,Platform,ChannelID,OriginalStatDate,RetainedUserCount)select A.SoftID,A.PLATFORM,A.FirstChannelID,B.OriginalEndDate,COUNT(distinct A.IMEI)from (select * from ....' + CAST((@StartDate / 10000) as nvarchar(10)) + N' with(nolock) where Part=@part and LoginDate between @StartDate and @EndDate and (FromCache=0 or FromCache is null)union allselect * from ....' + CAST((@EndDate / 10000) as nvarchar(10)) + N' with(nolock) where Part=@part and LoginDate between @StartDate and @EndDate and (FromCache=0 or FromCache is null)) Ainner join @dt Bon A.FirstLoginDate between B.OriginalStartDate and B.OriginalEndDategroup by A.SoftID,A.PLATFORM,A.FirstChannelID,B.OriginalEndDate;'end  declare @part tinyint = 0;    while @part < 128 begin      --if (@Period <> 12 or @part = 0) beginexec sp_executesql @sql, N'@part tinyint,@StartDate int,@EndDate int,@dt OriginalDatesType readonly', @Part, @StartDate, @EndDate, @dt;--endset @part = @part + 1endendselect @Period Period,OriginalStatDate,@EndDate StatDate,SoftID,Platform,-1 ID2,-1 ID1,0 OriginalNewUserCount,SUM(RetainedUserCount) RetainedUserCount from #RetainedUsersgroup by SoftID,Platform,OriginalStatDateselect @Period Period,OriginalStatDate,@EndDate StatDate,SoftID,Platform,ChannelID ID2,-1 ID1,0 OriginalNewUserCount,SUM(RetainedUserCount) RetainedUserCount from #RetainedUsersgroup by SoftID,Platform,ChannelID,OriginalStatDate  drop table #RetainedUsers
end

hadoop

@MapConfig
public static class MapTask extends Mapper<LongWritable, Text, Text, Text> {private Text mKey = new Text();private Text mValue = new Text();private StringBuilder sb = new StringBuilder();private Map<String,Integer> map=new HashMap<String,Integer>();@Overrideprotected void setup(Context context) throws IOException, InterruptedException {String enddate =context.getConfiguration().get("key_enddate");int period =Integer.parseInt(context.getConfiguration().get("period"));int step=Integer.parseInt(context.getConfiguration().get("step"));DateTime curstatdate=DateTime.parseToDateTime(enddate,"yyyyMMdd");//设置要计算留存的时间if (period==PeriodOptions.GetValueByEnum(PeriodOptions.Daily)){for (DateTime startdate=curstatdate.addDays(-30);new Double(DateTime.minusDay(curstatdate,startdate)).intValue()>0;startdate=startdate.addDays(1)){Integer tmp=Integer.parseInt(startdate.toString("yyyyMMdd"));map.put(startdate.toString("yyyyMMdd"),tmp);}}else if(period==PeriodOptions.GetValueByEnum(PeriodOptions.Weekly)){for (DateTime startdate=curstatdate.addDays(-48);DateTime.minusDay(curstatdate,startdate)>0;startdate=startdate.addDays(1)){if (WeekOptions.GetEnumByValue(startdate.getDayOfWeek())== WeekOptions.SUNDAY){Integer tmp=Integer.parseInt(startdate.toString("yyyyMMdd"));for (DateTime substartdate=startdate.addDays(-step);new Double(DateTime.minusDay(startdate,substartdate)).intValue()>=0;substartdate=substartdate.addDays(1)){map.put(substartdate.toString("yyyyMMdd"),tmp);}}}}else if(period==PeriodOptions.GetValueByEnum(PeriodOptions.NaturalMonth)){for (DateTime startdate=curstatdate.addMonths(-7).addDays(1);DateTime.minusDay(curstatdate,startdate)>0;startdate=startdate.addDays(1)){if (startdate.addDays(1).day()==1){Integer tmp=Integer.parseInt(startdate.toString("yyyyMMdd"));for (DateTime substartdate=startdate.addMonths(-1).addDays(1);new Double(DateTime.minusDay(startdate,substartdate)).intValue()>0;substartdate=substartdate.addDays(1)){map.put(substartdate.toString("yyyyMMdd"),tmp);}}}}}@Overrideprotected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {String str = value.toString();String[] params = StringUtils.splitByWholeSeparatorPreserveAllTokens(str, "\t");if (!map.containsKey(params[23]) || (Integer.parseInt(params[15])&1)!=0 ){return;}int firstlogintime=map.get(params[23]);String enddate =context.getConfiguration().get("key_enddate");sb.delete(0, sb.length());//key<0:softid,2:platform,22 firstchannelid>//value<23 firstlogindate,4:logindate,3:imei>sb.append(params[0]).append("\t").append(params[2]).append("\t").append(params[22]);mKey.set(sb.toString());mValue.set(firstlogintime + "\t" + enddate + "\t" + params[3]);context.write(mKey, mValue);sb.delete(0, sb.length());sb.append(params[0]).append("\t").append(params[2]).append("\t").append(-1);mKey.set(sb.toString());mValue.set(firstlogintime + "\t" + enddate + "\t" + params[3]);context.write(mKey, mValue);}@Overrideprotected void cleanup(Context context) throws IOException, InterruptedException {map.clear();}
}//key<0:softid,2:platform,22 firstchannelid>
//value<23 firstlogindate,4:logindate,3:imei>
@CombineConfig
public static class CombineTask extends Reducer<Text, Text, Text, Text> {Text mvalue=new Text();//留存用户private Multiset<String> multiset=HashMultiset.create();@Overrideprotected void reduce(Text key, Iterable<Text> values, Context context)throws IOException, InterruptedException {multiset.clear();for (Text item:values){multiset.add(item.toString());}for (String item:multiset.elementSet()){mvalue.set(item);context.write(key,mvalue);}}}//key<0:softid,2:platform,22 firstchannelid>
//value<23 firstlogindate,4:logindate,3:imei>
@ReduceConfig
public static class ReduceTask extends Reducer<Text, Text, Text, Text> {private Text mValue = new Text();//留存用户private Map<String, Multiset<Object>> MRetained = new HashMap<>();@Overrideprotected void reduce(Text key, Iterable<Text> values, Context context)throws IOException, InterruptedException {for (Map.Entry<String, Multiset<Object>> map : MRetained.entrySet()) {MRetained.get(map.getKey()).clear();}MRetained.clear();for (Text item : values) {String[] params = StringUtils.splitByWholeSeparatorPreserveAllTokens(item.toString(), "\t");String mapkey = params[0] + "\t" + params[1];if (!params[0].equals(params[1])) {if (!MRetained.containsKey(mapkey)) {MRetained.put(mapkey, HashMultiset.create());}MRetained.get(mapkey).add(params[2]);}}String period = context.getConfiguration().get("period");for (Map.Entry<String, Multiset<Object>> map : MRetained.entrySet()) {if (map.getKey().split("\t") == null ) {continue;}mValue.set(period+ "\t" + map.getKey()+ "\t" + 0+ "\t" + map.getValue().elementSet().size());//softid,platform,channelid,period,originaldate，statdate,OriginalNewUserCount,RetainedUserCountcontext.write(key, mValue);}}}

hive

insert overwrite table temp_db.temp_U_StatRetainedUsers partition(p=${period},dt='${dir}')
select a.soft_id,a.platform,a.channel,a.soft_version,a.country,${period},a.first_login_date,${date},a.newusercount,COALESCE(b.retainedusercount,0)
from (select soft_id,platform,channel,soft_version,country,count(1) newusercount,case when ${period}=1 then login_datewhen ${period}=3 then cast(from_unixtime(unix_timestamp(date_sub('${originalenddate2}',cast(floor(datediff('${originalenddate2}',from_unixtime(unix_timestamp(cast(login_date as string),'yyyyMMdd'),'yyyy-MM-dd'))/7)*7 as int)),'yyyy-MM-dd'),'yyyyMMdd') as int)when ${period}=12 then cast(from_unixtime(unix_timestamp(date_sub(from_unixtime(unix_timestamp(cast((floor(cast(from_unixtime(unix_timestamp(date_add(from_unixtime(unix_timestamp(cast((floor(login_date/100)*100+1) as string),'yyyyMMdd'),'yyyy-MM-dd'),31),'yyyy-MM-dd'),'yyyyMMdd') as int)/100)*100+1) as string),'yyyyMMdd'),'yyyy-MM-dd'),1),'yyyy-MM-dd'),'yyyyMMdd') as int) end first_login_datefrom computing_db.softloginuserswhere dt>='${datestart2}' and dt<='${dateend2}'group by soft_id,platform,channel,soft_version,country,case when ${period}=1 then login_datewhen ${period}=3 then cast(from_unixtime(unix_timestamp(date_sub('${originalenddate2}',cast(floor(datediff('${originalenddate2}',from_unixtime(unix_timestamp(cast(login_date as string),'yyyyMMdd'),'yyyy-MM-dd'))/7)*7 as int)),'yyyy-MM-dd'),'yyyyMMdd') as int)when ${period}=12 then cast(from_unixtime(unix_timestamp(date_sub(from_unixtime(unix_timestamp(cast((floor(cast(from_unixtime(unix_timestamp(date_add(from_unixtime(unix_timestamp(cast((floor(login_date/100)*100+1) as string),'yyyyMMdd'),'yyyy-MM-dd'),31),'yyyy-MM-dd'),'yyyyMMdd') as int)/100)*100+1) as string),'yyyyMMdd'),'yyyy-MM-dd'),1),'yyyy-MM-dd'),'yyyyMMdd') as int) end) a
left outer join (select soft_id,platform,first_channel,first_soft_version,first_country,count(distinct dev_id) retainedusercount,case when ${period}=1 then first_login_datewhen ${period}=3 then cast(from_unixtime(unix_timestamp(date_sub('${originalenddate2}',cast(floor(datediff('${originalenddate2}',from_unixtime(unix_timestamp(cast(first_login_date as string),'yyyyMMdd'),'yyyy-MM-dd'))/7)*7 as int)),'yyyy-MM-dd'),'yyyyMMdd') as int)when ${period}=12 then cast(from_unixtime(unix_timestamp(date_sub(from_unixtime(unix_timestamp(cast((floor(cast(from_unixtime(unix_timestamp(date_add(from_unixtime(unix_timestamp(cast((floor(first_login_date/100)*100+1) as string),'yyyyMMdd'),'yyyy-MM-dd'),31),'yyyy-MM-dd'),'yyyyMMdd') as int)/100)*100+1) as string),'yyyyMMdd'),'yyyy-MM-dd'),1),'yyyy-MM-dd'),'yyyyMMdd') as int) end first_login_datefrom computing_db.softloginlogwhere dt>='${datestart}' and dt<='${dateend}' and first_login_date between ${originalstartdate} and ${originalenddate} and from_cache=0group by soft_id,platform,first_channel,first_soft_version,first_country,case when ${period}=1 then first_login_datewhen ${period}=3 then cast(from_unixtime(unix_timestamp(date_sub('${originalenddate2}',cast(floor(datediff('${originalenddate2}',from_unixtime(unix_timestamp(cast(first_login_date as string),'yyyyMMdd'),'yyyy-MM-dd'))/7)*7 as int)),'yyyy-MM-dd'),'yyyyMMdd') as int)when ${period}=12 then cast(from_unixtime(unix_timestamp(date_sub(from_unixtime(unix_timestamp(cast((floor(cast(from_unixtime(unix_timestamp(date_add(from_unixtime(unix_timestamp(cast((floor(first_login_date/100)*100+1) as string),'yyyyMMdd'),'yyyy-MM-dd'),31),'yyyy-MM-dd'),'yyyyMMdd') as int)/100)*100+1) as string),'yyyyMMdd'),'yyyy-MM-dd'),1),'yyyy-MM-dd'),'yyyyMMdd') as int) end) b
on a.soft_id=b.soft_id and a.platform=b.platform and a.channel=b.first_channel and a.country=b.first_country and a.soft_version=b.first_soft_version and a.first_login_date=b.first_login_date;