Spark RDD案例：统计每日新增用户
Posted howard2005
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了Spark RDD案例：统计每日新增用户相关的知识，希望对你有一定的参考价值。
文章目录
一、提出任务
- 已知有以下用户访问历史数据，第一列为用户访问网站的日期，第二列为用户名：
2022-01-01,mike
2022-01-01,alice
2022-01-01,brown
2022-01-02,mike
2022-01-02,alice
2022-01-02,green
2022-01-03,alice
2022-01-03,smith
2022-01-03,brian
| 2022-01-01 | mike | alice | brown |
| 2022-01-02 | mike | alice | green |
| 2022-01-03 | alice | smith | brian |
- 现需要根据上述数据统计每日新增的用户数量，期望统计结果：
2022-01-01,3
2022-01-02,1
2022-01-03,2
- 即2022-01-01新增了3个用户（分别为mike、alice、brown），2022-01-02新增了1个用户（green），2022-01-03新增了两个用户（分别为smith、brian）。
二、实现思路
- 使用倒排索引法，若将用户名看作关键词，访问日期看作文档ID，则用户名与访问日期的映射关系如下图所示。
| 2022-01-01 | 2022-01-02 | 2022-01-3 | |
|---|---|---|---|
| mike | √ | √ | |
| alice | √ | √ | √ |
| brown | √ | ||
| green | √ | ||
| smith | √ | ||
| brian | √ |
- 若同一个用户对应多个访问日期，则最小的日期为该用户的注册日期，即新增日期，其他日期为重复访问日期，不应统计在内。因此每个用户应该只计算用户访问的最小日期即可。如下图所示，将每个用户访问的最小日期都移到第一列，第一列为有效数据，只统计第一列中每个日期的出现次数，即为对应日期的新增用户数。
| 列一 | 列二 | 列三 | |
|---|---|---|---|
| mike | 2022-01-01 | 2022-01-02 | |
| alice | 2022-01-01 | 2022-01-02 | 2022-01-03 |
| brown | 2022-01-01 | ||
| green | 2022-01-02 | ||
| smith | 2022-01-03 | ||
| brian | 2022-01-03 |
- 预备工作：启动集群的HDFS与Spark
- 在HDFS上准备数据 -
users.txt
三、完成任务
（一）读取文件，得到RDD
- 执行命令：
val rdd1 = sc.textFile("hdfs://master:9000/input/users.txt")
（二）倒排，互换RDD中元组的元素顺序
val rdd2 = rdd1.map(
line =>
val fields = line.split(",")
(fields(1), fields(0))
)
- 执行上述语句
（三）倒排后的RDD按键分组
- 执行命令：
val rdd3 = rdd2.groupByKey()
（四）取分组后的日期集合最小值，计数为1
- 执行命令：
val rdd4 = rdd3.map(line => (line._2.min, 1))
（五）按键计数，得到每日新增用户数
- 执行命令：
val result = rdd4.countByKey() - 执行命令：
result.keys.foreach(key => println(key + "," + result(key)))
（六）让输出结果按日期升序
- 映射不能直接排序，只能让键集转成列表之后先排序，再遍历键集输出映射
- 执行命令：
val keys = result.keys.toList.sorted，让键集升序排列
四、课后作业
- 创建Maven项目 -
CountNewUsers，完成每日新增用户统计任务
以上是关于Spark RDD案例：统计每日新增用户的主要内容，如果未能解决你的问题，请参考以下文章