一個統計用戶新增活躍留存的方案

2022 年 2 月 10 日
筆記
方案

背景

產品上線後，出於運營的需要，我們要對用戶進行跟蹤，分析用戶數據。本文要介紹的是如何統計用戶新增數、活躍數和留存率，時間跨度是天，即統計每日新增（DNU），日活（DAU）和某日新增的一批用戶在接下來的一段時間內每天活躍的百分比。

使用範圍

本方案適用於用戶量不太大（日活在百萬以內，日活百萬以上不是不能用，只是在統計數據時耗時太長不太合適），尤其適合小團隊或個人開發者（比如你公司服務端接口開發是你，運維也是你，現在老闆又來叫你做數據報表）。如果你的產品的日活有幾百萬甚至幾千萬或過億，這樣的產品當然是完全可以養一個大數據部門的，本方案並不適用這種情況。

涉及到的工具和技術點

shell腳本

本方案需要你懂一點兒shell，起碼能看懂，也要求你知道怎麼寫crontab定時任務。

MySQL

本方案需要你熟練使用sql，知道怎麼定義存儲過程，知道分區表的概念和用法。

實現過程

一、目標

由於數據量是不斷增加的，所以我們的目標是要把原始數據聚合成一張可以直接用一條select語句就可以查看每日新增、日活和留存率的表，並且只能做單表查詢，否則當數據量增大時，聯表查詢的速度會大大下降。而且為了防止出錯，我們的數據還需要可以重跑但是不會影響到已存在的數據。

最終呈現給運營人員看到的數據是這樣的：

新增-活躍表

日期	維度1	維度2	新增數	活躍數
2022-02-09	合計	合計	1000	5000
2022-02-08	合計	合計	1000	4500
2022-02-07	合計	合計	1000	4000

用戶留存表

日期	維度1	維度2	新增數	次日留存率	2日留存率
2022-02-09	合計	合計	1000	0	0
2022-02-08	合計	合計	1000	60%	0
2022-02-07	合計	合計	1000	60%	50%

簡單解釋一下上面兩個表的結構：因為我們是按天統計的，所以日期都是以天為單位，用戶可能有不同的國家或地區，不同版本，不同手機型號等等，所以就有了各個維度。用戶留存表的數據要注意，比如今天是2022年2月9日，那麼就只能統計到9號的新增，9號新增用戶的日次留存是10號才能統計到的，但是8號新增用戶的次日留存在今天（也就是9號）就統計出來了，所以留存的數據是一個階梯形狀的。

二、收集數據

為了方便介紹本方案，這裡假設只有日期、國家、版本號三個維度。

收集數據的下一步是數據入庫，為了方便，需要把數據格式進行轉換。因為服務端接口現在一般都是使用json格式的數據進行通信，如果直接把json格式的數據輸出到日誌文件，處理起來會非常麻煩，所以需要在服務端接收到統計日誌時，把數據輸出到單獨的日誌文件中，還要按照MySQL的load命令可以識別的數據格式。

在輸出日誌之前，先確定好都需要哪些數據，這裡需要的數據如下：

ts：timestamp，時間戳。服務端接收到日誌的時間，格式是yyyy-MM-DD HH:mm:ss。

device_id：設備id，這裡是用來唯一標識用戶的一個字符串，比如在android設備上可以用android id，總之這個字段是用來確定一台設備的，要保證不同的設備設備id不同。

country：用戶所在的國家。如果你是只做一個國家的，比如只做國內市場，也可以把這個字段換成省份或者城市，總之根據運營需求去改變。

version：應用版本號，一般是一個整數。

於是就可以確定日誌的格式如下：

2022-02-09 13:14:15||aaaaaa||CN||100
2022-02-09 13:14:16||bbbbbb||US||100
2022-02-09 13:14:17||cccccc||NL||100

也就是一條數據佔一行，字段之間使用雙豎線分隔，當然這裡不一定是雙豎線分隔，也可以換成其它的，原則是字符數少而且不能被字段的值包含，不然在數據入庫時會出現字段不對應的問題。

再考慮兩個方面：

如果數據量較大要怎麼處理？
可能有的字段的長度沒法一下子確定怎麼處理？
保留數據的策略應該怎樣設置？

第1個問題，當數據量大時，可以考慮把日誌文件切割成更小的時間段，比如每小時一個日誌文件，然後下一小時就把上一個小時的數據入庫。

第2個問題，原始數據表的字段長度定義得大一些，做到即使以後字段有變化，也可以適應。

第3個問題，因為我們的目標是跑出最後的報表，所以不可能一直保存着所有的原始日誌數據，為了防止出錯，可能只是保留最近幾天的，一個簡單的策略是在每次日誌數據入庫前用delete語句把前幾天的數據刪除了，但是直接使用delete有兩個問題：一是MySQL要掃描全表刪除數據，比較耗時；而是MySQL的delete + where刪除可能只是假刪除，磁盤不會立即釋放。所以這裡使用分區表來實現，每天的數據作為一個分區，刪除數據時直接刪除分區，數據入庫時先創建分區。

於是得到原始數據表的DDL如下：

CREATE TABLE `st_base` (
  `id` bigint(20) NOT NULL AUTO_INCREMENT COMMENT 'ID',
  `dd` int(11) NOT NULL DEFAULT '0' COMMENT '天數，格式是yyyyMMddHH',
  `ts` varchar(64) COLLATE utf8_bin NOT NULL DEFAULT '' COMMENT '時間戳',
	`device_id` varchar(64) COLLATE utf8_bin NOT NULL DEFAULT '' COMMENT '設備id',
	`country` varchar(64) COLLATE utf8_bin NOT NULL DEFAULT '' COMMENT '國家',
	`version` varchar(64) COLLATE utf8_bin NOT NULL DEFAULT '' COMMENT '版本號',
	PRIMARY KEY (`id`,`dd`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8 COLLATE=utf8_bin COMMENT='原始數據表'
/*!50100 PARTITION BY LIST (dd)
(PARTITION p20220209 VALUES IN (20220209) ENGINE = InnoDB) */

三、數據入庫

有了格式化的日誌文件和數據表，就可以通過shell腳本把數據入庫了。步驟如下：

刪除歷史日誌的分區
刪除執行日期的分區（這一步在重跑數據很有用）
創建執行日期的分區
使用MySQL的load命令把數據從日誌文件加載到數據庫中

這裡只說一下重要的命令：

刪除和創建分區可以分別使用下面兩個命令：

drop_sql="alter table st_base drop partition pxxxxxxxx" # 這裡的xxxxxxxx要根據執行日期轉換一下
add_sql="alter table st_base add partition (partition pxxxxxxxx values in (xxxxxxxx) engine=innodb)"

mysql -u${username} -p${password} -D${database} -e "${drop_sql}"

mysql -u${username} -p${password} -D${database} -e "${add_sql}"

上面使用mysql命令指定了用戶名、密碼、數據庫名和sql語句（-e參數）

從文件加載數據入庫

log_file=xxxx #日誌文件名
dd=xxxxxxxx #執行日期
load_sql="load data local infile '${log_file}' ignore into table st_base fields terminated by '||' lines terminated by '\n' (ts,device_id,country,version) set dd='${dd}'"
mysql -u${username} -p${password} -D${database} -e "${load_sql}"

定時任務

因為我們是每天入庫一次，所以可以在每天的0時10分去跑上面的腳本任務。假設上面的腳本文件保存為st_base.sh

可以通過crontab -e編輯定時任務：

10 0 * * * /path/to/job/st_base.sh

當然最好的做法是把執行日期當做腳本的參數傳入，這樣可以實現重跑某天的數據了。

四、清洗數據

在上一步得到了原始數據之後，接下來的工作都可以在MySQL中完成，首先要清洗數據。

這一步的目的有兩個：

確定好數據類型
數據去重

先創建一個臨時表tmp_base，這個表用來轉換數據類型，如果有一些字段的值需要轉換的也可以在這裡做（舉個例子：假如客戶端獲取到的國家有幾種途徑，分別是獲取了sim卡國家，網絡國家，手機國家，到了服務端後服務器根據客戶端的ip也解析出了一個國家，但是運營的時候可能只需要一個最接近用戶的真實國家，那麼就可以按照優先級來確定，當然本文沒有多個國家的問題），DDL如下：

CREATE TABLE `tmp_base` (
  `id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'ID',
  `dt` date NOT NULL COMMENT '日期',
	`device_id` varchar(32) COLLATE utf8_bin NOT NULL DEFAULT '' COMMENT '設備id',
	`country` varchar(8) COLLATE utf8_bin DEFAULT NULL,
  `version` int(11) NOT NULL DEFAULT '0' COMMENT '版本號',
  PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8 COLLATE=utf8_bin COMMENT='用戶基礎臨時表'

再創建一個用戶總表total_base，這個表用來存放所有用戶的數據，每個用戶只有一條數據，DDL如下：

CREATE TABLE `total_base` (
  `id` bigint(20) NOT NULL AUTO_INCREMENT COMMENT 'ID',
  `dt` date NOT NULL COMMENT '新增日期',
  `device_id` varchar(32) COLLATE utf8_bin NOT NULL DEFAULT '' COMMENT '設備id',
	`country` varchar(8) COLLATE utf8_bin DEFAULT NULL,
  `version` int(11) NOT NULL DEFAULT '0' COMMENT '版本號',
  PRIMARY KEY (`id`),
  UNIQUE KEY `device` (`device_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8 COLLATE=utf8_bin COMMENT='用戶總表';

創建一個流水表flow_base，同樣以日期作為分區字段，DDL如下：

CREATE TABLE `flow_base` (
  `id` bigint(20) NOT NULL AUTO_INCREMENT COMMENT 'ID',
  `dt` date NOT NULL DEFAULT '2022-01-01' COMMENT '日期',
	`device_id` varchar(32) COLLATE utf8_bin NOT NULL DEFAULT '' COMMENT '設備id',
	`country` varchar(8) COLLATE utf8_bin DEFAULT NULL,
  `version` int(11) NOT NULL DEFAULT '0' COMMENT '版本號',
  `rdt` date NOT NULL DEFAULT '2022-01-01' COMMENT '用戶註冊日期',
  `dd` int(11) NOT NULL DEFAULT '0' COMMENT '日期(yyyyMMdd)，用來做分區',
  PRIMARY KEY (`id`,`dd`),
  UNIQUE KEY `unique` (`dt`,`device_id`,`dd`)
) ENGINE=InnoDB CHARSET=utf8 COLLATE=utf8_bin COMMENT='用戶基礎流水表'
/*!50100 PARTITION BY LIST (dd)
(PARTITION p20220209 VALUES IN (20220209) ENGINE = InnoDB) */

注意到流水表flow_base中有一個rdt的字段，這字段是用來存放這個用戶的註冊日期，方便後面統計留存使用的。

準備好表結構之後，開始清洗數據。清洗數據使用MySQL的存儲過程功能，創建一個存儲過程sp_data_cleaning，在這個存儲過程中，需要做以下幾件事：

把原始數據表st_base中的數據清洗到臨時表tmp_base，如果有字段的值需要轉換也在這一步做。
把臨時表tmp_base中的用戶添加到用戶總表total_base中。
把臨時表tmp_base中的數據添加到流水表中，並且聯合用戶總表，把用戶的註冊日期也填充好。

於是可以得到存儲過程sp_data_cleaning的DDL如下：

CREATE PROCEDURE `sp_data_cleaning`(IN v_dt VARCHAR(10))
BEGIN
	# 變量
	declare pname varchar(10);
	declare v_is_pname_exists int;
	
	# 清除tmp_base表數據
    truncate table tmp_base;

	# 清洗數據
		insert into tmp_base(
			`dt`,
	  	`device_id`,
			`country`,
	  	`version`
			)
		select
			v_dt,
			`device_id`,
	  	`country`,
	  	`version`
	  from `st_base`
  	where `dd` = replace(v_dt,'-','');
  	
	# 數據加入用戶總表
		insert ignore into total_base(
			`dt`,
			`device_id`,
			`country`,
			`version`
			)
		select
			`dt`,
			`device_id`,
			`country`,
			`version`
		from tmp_base;
  
	# 給流水表創建分區
	select concat('p', replace(v_dt, '-', '')) into pname;
	
	# 查找是否已經存在執行日期的分區
	select max(a) into v_is_pname_exists
	from (
		select 1 as a from information_schema.PARTITIONS 
		where `TABLE_SCHEMA` = 'your_database_name' 
			and `TABLE_NAME` = 'flow_base'
			and `PARTITION_NAME`=pname
		union all
		select 0 
		) t;
	
	# 如果已經存在先刪除
	if v_is_pname_exists=1 then
	set @drop_sql=concat('alter table flow_base drop partition ', pname);
	prepare stmt from @drop_sql;
	execute stmt;
	deallocate prepare stmt;
	end if;

	# 創建分區
	set @add_sql=concat('alter table flow_base add partition (partition ', pname, ' values in (', v_date, ') ENGINE = InnoDB)');
	prepare stmt from @add_sql;
	execute stmt;
	deallocate prepare stmt;		

	# 數據加入流水表
		insert ignore into flow_base(
			`dt`,
			`device_id`,
			`country`,
			`version`,
			`rdt`,
			`dd`
			)
		select
			v_dt,
			t1.`device_id`,
			t1.`country`,
			t1.`version`,
			t2.`dt`,
			replace(v_dt, '-', '')
		from tmp_base t1
		left outer join total_base t2
		on (t1.`device_id`=t2.`device_id`);
END

五、數據聚合

經過上面幾個步驟的處理，現在已經得到了半成品的數據，可以進行聚合了。根據第一步的目標報表，可以確定兩個表的結構：一個是用戶的新增-活躍表，另一個是用戶的留存表，DDL如下：

新增-活躍表：

CREATE TABLE `rpt_base_active` (
  `id` bigint(20) NOT NULL AUTO_INCREMENT COMMENT 'ID',
  `dt` date NOT NULL DEFAULT '2022-01-01' COMMENT '日期',
	`country` varchar(8) COLLATE utf8_bin DEFAULT NULL,
  `version` int(11) NOT NULL DEFAULT '0' COMMENT '版本號',
  `new_users` smallint(4) NOT NULL DEFAULT '0' COMMENT '新增數',
  `active_users` smallint(4) NOT NULL DEFAULT '0' COMMENT '活躍數',
  PRIMARY KEY (`id`),
  KEY `index1` (`dt`),
  KEY `index3` (`country`,`version`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8 COLLATE=utf8_bin COMMENT='用戶新增活躍表'

用戶留存表（這裡假設只看7天的留存情況，如果需要看更多留存天數，可以自行修改）：

CREATE TABLE `rpt_base` (
  `id` bigint(20) NOT NULL AUTO_INCREMENT COMMENT 'ID',
  `dt` date NOT NULL DEFAULT '2022-01-01' COMMENT '日期',
	`country` varchar(8) COLLATE utf8_bin DEFAULT NULL,
  `version` int(11) NOT NULL DEFAULT '0' COMMENT '版本號',
  `d0` smallint(4) NOT NULL DEFAULT '0' COMMENT '新增數',
  `d1` smallint(4) NOT NULL DEFAULT '0' COMMENT '次日留存數',
  `d2` smallint(4) NOT NULL DEFAULT '0' COMMENT '2日留存數',
  `d3` smallint(4) NOT NULL DEFAULT '0' COMMENT '3日留存數',
  `d4` smallint(4) NOT NULL DEFAULT '0' COMMENT '4日留存數',
  `d5` smallint(4) NOT NULL DEFAULT '0' COMMENT '5日留存數',
  `d6` smallint(4) NOT NULL DEFAULT '0' COMMENT '6日留存數',
  `d7` smallint(4) NOT NULL DEFAULT '0' COMMENT '7日留存數',
  PRIMARY KEY (`id`),
  KEY `index1` (`dt`),
  KEY `index3` (`country`,`version`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8 COLLATE=utf8_bin COMMENT='用戶留存表'

注意，以上兩個表的索引創建並不是固定的，需要根據運營的實際情況去創建相關的索引。

在跑數據之前，先聚合一下執行日期的數據，創建一個臨時表a_flow_base，這個表用來初步聚合數據，DDL如下：

CREATE TABLE `a_flow_base` (
  `id` bigint(20) NOT NULL AUTO_INCREMENT COMMENT 'ID',
  `dt` date NOT NULL DEFAULT '2022-01-01' COMMENT '日期',
	`country` varchar(8) COLLATE utf8_bin DEFAULT NULL,
  `version` int(11) NOT NULL DEFAULT '0' COMMENT '應用版本號',
  `rdt` date NOT NULL DEFAULT '2022-01-01' COMMENT '用戶註冊日期',
  `rdays` smallint(4) NOT NULL DEFAULT '0' COMMENT '留存天數',
  `users` smallint(4) NOT NULL DEFAULT '0' COMMENT '用戶數',
  PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8 COLLATE=utf8_bin COMMENT='用戶基礎數據聚合表'

首先初步聚合用戶數據，創建一個存儲過程sp_a_flow_base，DDL如下：

CREATE PROCEDURE `sp_a_flow_base`(in v_dt char(10))
BEGIN
	declare d0 date;
	declare d1 date;
	declare d2 date;
	declare d3 date;
	declare d4 date;
	declare d5 date;
	declare d6 date;
	declare d7 date;
	
	select date_sub(v_dt, interval 0 day) into d0;
	select date_sub(v_dt, interval 1 day) into d1;
	select date_sub(v_dt, interval 2 day) into d2;
	select date_sub(v_dt, interval 3 day) into d3;
	select date_sub(v_dt, interval 4 day) into d4;
	select date_sub(v_dt, interval 5 day) into d5;
	select date_sub(v_dt, interval 6 day) into d6;
	select date_sub(v_dt, interval 7 day) into d7;
	
	# 清除a_flow_base表數據
	truncate table a_flow_base;
	
	insert into a_flow_base(
		`dt`,
		`country`,
  	`version_code`,
  	`rdt`,
  	`rdays`,
  	`users`
		)
	select 
		t1.`dt`,
  	t1.`country`,
  	t1.`version`,
  	t1.`rdt`,
  	datediff(t1.`dt`, t1.`rdt`) as rdays,
  	count(*) as users
	from flow_base t1
	where t1.`dt` in (d0,d1,d2,d3,d4,d5,d6,d7)
	group by 
		t1.`dt`,
  	t1.`country`,
  	t1.`version`,
  	t1.`rdt`;
END

初步聚合了數據後，開始正式聚合數據，創建一個存儲過程sp_rpt_base，DDL如下：

CREATE PROCEDURE `sp_rpt_base`(in v_dt char(10))
BEGIN
	declare d0 date;
	declare d1 date;
	declare d2 date;
	declare d3 date;
	declare d4 date;
	declare d5 date;
	declare d6 date;
	declare d7 date;
	
	select date_sub(v_dt, interval 0 day) into d0;
	select date_sub(v_dt, interval 1 day) into d1;
	select date_sub(v_dt, interval 2 day) into d2;
	select date_sub(v_dt, interval 3 day) into d3;
	select date_sub(v_dt, interval 4 day) into d4;
	select date_sub(v_dt, interval 5 day) into d5;
	select date_sub(v_dt, interval 6 day) into d6;
	select date_sub(v_dt, interval 7 day) into d7;
	
	# 刪除數據
	delete from rpt_base_active where `dt` = v_dt;
	
	insert into rpt_base_active (
		`dt`,
		`country`,
  	`version`,
  	`new_users`,
  	`active_users`
		)
	select
		`dt`,
		`country`,
		`version`,
		sum(if(`dt`=`rdt`, 1, 0)) as `new_users`,
		sum(1) as `active_users`
	from flow_base
	where dt=v_dt
	group by 
		`dt`,
		`country`,
		`version`
		;

	# 刪除數據
	delete from rpt_base where `dt` in (d0,d1,d2,d3,d4,d5,d6,d7);
	
	insert into rpt_base(
		`dt`,
		`country`,
  	`version`,
  	`d0`,
  	`d1`,
  	`d2`,
  	`d3`,
  	`d4`,
  	`d5`,
  	`d6`,
  	`d7`
		)
	select
		t1.`rdt`,
		t1.`country`,
		t1.`version`,
		sum(case when t1.`rdays`=0 then t1.`users` else 0 end) as d0,
		sum(case when t1.`rdays`=1 then t1.`users` else 0 end) as d1,
		sum(case when t1.`rdays`=2 then t1.`users` else 0 end) as d2,
		sum(case when t1.`rdays`=3 then t1.`users` else 0 end) as d3,
		sum(case when t1.`rdays`=4 then t1.`users` else 0 end) as d4,
		sum(case when t1.`rdays`=5 then t1.`users` else 0 end) as d5,
		sum(case when t1.`rdays`=6 then t1.`users` else 0 end) as d6,
		sum(case when t1.`rdays`=7 then t1.`users` else 0 end) as d7
		from a_flow_base t1
		group by
			t1.`rdt`,
			t1.`country`,
  		t1.`version`
			;
			
END

為了方便調用整個過程，可以再創建一個存儲過程，把全過程寫在一起，一次執行。創建一個存儲過程sp_user，DDL如下：

CREATE PROCEDURE `sp_user`(in v_dt char(10))
BEGIN

	call sp_tmp_base(v_dt);
	call sp_data_cleaning(v_dt);
	call sp_a_flow_base(v_dt);
	call sp_rpt_base(v_dt);
	
END

這樣，就可以添加定時任務每天定時跑前一天的數據了。

寫在最後

流水表flow_base應該保留幾天的數據？

這個看你的用戶留存表需要看多少天留存數據，如果你要看7日留存，那麼保留最近8天的數據，如果是想看30天留存，就保留最近31天的數據，依次類推。

如果運營人員或老闆9點半上班，每天凌晨的0點開始跑前一天的數據，你將有9個半小時來跑前一天的數據。當然如果一天的數據要跑2個小時以上，還是考慮用Hadoop來做吧。“

Tags: 方案