揭秘：如何精准统计连续登录三天及以上的用户？

人亦已歌

揭秘：如何精准统计连续登录三天及以上的用户？

[TOC] ### 一、需求 > 统计连续登录三天及以上的用户 + 这个问题可以扩展到很多相似的问题：连续几个月充值会员、连续天数有商品卖出、连续打滴滴、连续逾期。 示例 | uid | times | start_date | end_date | | ------ | ----- | ---------- | ---------- | | guid01 | 4 | 2018-03-04 | 2018-03-07 | | guid02 | 3 | 2018-03-01 | 2018-03-03 | ### 二、数据准备 + `v_user_login.csv`

2. uid,datatime
3. guid01,2018-02-28
4. guid01,2018-03-01
5. guid01,2018-03-02
6. guid01,2018-03-04
7. guid01,2018-03-05
8. guid01,2018-03-06
9. guid01,2018-03-07
10. guid02,2018-03-01
11. guid02,2018-03-02
12. guid02,2018-03-03
13. guid02,2018-03-06

复制代码

说明 + uid:登录用户ID + datatime:登录时间 ### 三、SQL实现 #### 3.1 步骤 1.查询用户登录详情，并且根据访问时间顺序对每个用户访问次数进行累计 2.查询出连续登录日期 - 思考：如何能够确定是否是连续时间登录 - 寻找规律：登录累积次数减登录时间，如果相同则代码是连续时间登录 3. 对uid和dif进行分组，统计数量 #### 3.2 代码

2. import org.apache.spark.sql.{DataFrame, SparkSession}
3. object UserContinueLoginSQL {
4. def main(args: Array[String]): Unit = {
5. val spark = SparkSession.builder().appName("UserContinueLoginSQL").master("local[*]").getOrCreate()
6. import spark.implicits._
7. val df: DataFrame = spark.read
8. .option("header", "true")
9. .option("inferSchema", "true")
10. .csv("data_sql/v_user_login.csv")
11. df.createOrReplaceTempView("v_access_log")
12. //1.查询用户登录详情，并且根据访问时间顺序对每个用户访问次数进行累计
13. //row_number()函数将针对SELECT语句返回的每一行，从1开始编号，赋予其连续的编号。
14. spark.sql(
15. """
16. | select
17. | uid,
18. | datatime,
19. | ROW_NUMBER() OVER(PARTITION BY uid ORDER BY datatime ASC) rn
20. | from v_access_log
21. """.stripMargin).createOrReplaceTempView("t1")
22. //2.查询出连续登录日期
23. //思考：如何能够确定是否是连续时间登录
24. //寻找规律：登录累积次数减登录时间，如果相同则代码是连续时间登录
25. //DATE_SUB():日期相减
26. spark.sql(
27. """
28. | select
29. | uid,
30. | datatime,
31. | DATE_SUB(datatime,rn) dif
32. | from t1
33. """.stripMargin).createOrReplaceTempView("t2")
34. //3.对uid和dif进行分组，统计数量
35. spark.sql(
36. """
37. | select
38. | uid,
39. | MIN(datatime) start_date,
40. | MAX(datatime) end_date,
41. | count(1) counts
42. | from t2
43. | group by uid,dif HAVING counts >= 3
44. """.stripMargin).show()
45. }
46. }

复制代码

运行结果 +------+----------+----------+------+ | uid|start_date| end_date|counts| +------+----------+----------+------+ |guid02|2018-03-01|2018-03-03| 3| |guid01|2018-02-28|2018-03-02| 3| |guid01|2018-03-04|2018-03-07| 4| +------+----------+----------+------+ ### 四、RDD实现

2. import java.text.SimpleDateFormat
3. import java.util.{Calendar, Date}
4. import org.apache.spark.rdd.RDD
5. import org.apache.spark.{SparkConf, SparkContext}
6. object UserContinueLoginRDD {
7. def main(args: Array[String]): Unit = {
8. val conf = new SparkConf().setAppName("UserContinueLoginRDD").setMaster("local[*]")
9. val sc = new SparkContext(conf)
10. //读取数据
11. val rdd: RDD[String] = sc.textFile("data_sql/v_user_login.csv")
12. //过滤掉第一行表头
13. val header = rdd.first()
14. val rdd1 = rdd.filter(row => row != header)
15. //对数据进行处理
16. val uidAndDate: RDD[(String, String)] = rdd1.map(x => {
17. val fis = x.split(",")
18. val uid = fis(0)
19. val date = fis(1)
20. (uid, date)
21. })
22. //根据uid进行分分组，将同一个用户的登录数据搞到同一个分组中
23. val grouped: RDD[(String, Iterable[String])] = uidAndDate.groupByKey()
24. //在组内进行排序
25. val uidAndDateDif = grouped.flatMapValues(it => {
26. //将迭代器中的数据toList/toSet，有可能会发生内存溢出
27. val sorted = it.toSet.toList.sorted
28. //定义一个日期的工具类
29. val calendar = Calendar.getInstance()
30. val sdf = new SimpleDateFormat("yyyy-MM-dd")
31. var index = 0;
32. sorted.map(desStr => {
33. val date: Date = sdf.parse(desStr)
34. calendar.setTime(date)
35. calendar.add(Calendar.DATE, -index)
36. index += 1
37. (desStr, sdf.format(calendar.getTime))
38. })
39. })
40. val result = uidAndDateDif.map(x => {
41. ((x._1, x._2._2), x._2._1)
42. }).groupByKey().mapValues(it=>{
43. val list = it.toList
44. val times = list.size
45. val startTime = list.head
46. val endTime = list.last
47. (times,startTime,endTime)
48. }).map(t=>{
49. (t._1._1,t._2._1,t._2._2,t._2._3)
50. }).filter(x=>{
51. x._2>=3
52. })
53. val buffer = result.collect().toBuffer
54. println(buffer)
55. }
56. }

复制代码

运行结果 ArrayBuffer((guid02,3,2018-03-01,2018-03-03), (guid01,4,2018-03-04,2018-03-07), (guid01,3,2018-02-28,2018-03-02))