2024年大数据最全大数据Spark实时搜索日志实时分析_百度搜索日志数据集(2)，做了三年大数据开发

既有适合小白学习的零基础资料，也有适合3年以上经验的小伙伴深入学习提升的进阶课程，涵盖了95%以上大数据知识点，真正体系化！

由于文件比较多，这里只是将部分目录截图出来，全套包含大厂面经、学习笔记、源码讲义、实战项目、大纲路线、讲解视频，并且后续会持续更新

需要这份系统化资料的朋友，可以戳这里获取

• 方法二：consumerKafka，消费Kafka Topic中数据
  具体代码如下：

实时从Kafka Topic消费数据，提取ip地址字段，调用【ip2Region】库解析为省份和城市，存储到HDFS文件中，设置批处理时间间隔BatchInterval为10秒，完整代码如下：

运行模拟日志数据程序和ETL应用程序，查看实时数据ETL后保存文件，截图如下：

实 时 累 加 统 计 用 户 各 个 搜 索 词 出 现 的 次 数 ， 在 SparkStreaming 中 提 供 函 数【updateStateByKey】实现累加统计，Spark 1.6提供【mapWithState】函数状态统计，性能更好，实际应用中也推荐使用。

4.1 updateStateByKey 函数

状态更新函数【updateStateByKey】表示依据Key更新状态，要求DStream中数据类型为【Key/Value】对二元组，函数声明如下：

将每批次数据状态，按照Key与以前状态，使用定义函数【updateFunc】进行更新，示意图如下：

文档： http://spark.apache.org/docs/2.4.5/streaming-programming-guide.html#updatestatebykey-operation
针对搜索词词频统计WordCount，状态更新逻辑示意图如下：
以前的状态数据，保存到Checkpoint检查点目录中，所以在代码中需要设置Checkpoint检查点目录：
完整演示代码如下：

运行应用程序，通过WEB UI界面可以发现，将以前状态保存到Checkpoint检查点目录中，更新时在读取。
此外，updateStateByKey函数有很多重载方法，依据不同业务需求选择合适的方式使用。

4.2 mapWithState 函数

Spark 1.6提供新的状态更新函数【mapWithState】，mapWithState函数也会统计全局的key的状态，但是如果没有数据输入，便不会返回之前的key的状态，只是关心那些已经发生的变化的key，对于没有数据输入，则不会返回那些没有变化的key的数据。
这样的话，即使数据量很大，checkpoint也不会像updateStateByKey那样，占用太多的存储，效率比较高；
需要构建StateSpec对象，对状态State进行封装，可以进行相关操作，类的声明定义如下：
状态函数【mapWithState】参数相关说明：
完整演示代码如下：

运行程序可以发现，当Key（搜索单词）没有出现时，不会更新状态，仅仅更新当前批次中出现的Key的状态。
mapWithState 实现有状态管理主要是通过两点：a）、历史状态需要在内存中维护，这里必需的了，updateStateBykey也是一样；b）、自定义更新状态的mappingFunction，这些就是具体的业务功能实现逻辑了(什么时候需要更新状态)
首先数据像水流一样从左侧的箭头流入，把mapWithState看成一个转换器的话，mappingFunc就是转换的规则，流入的新数据(key-value)结合历史状态(通过key从内存中获取的历史状态)进行一些自定义逻辑的更新等操作，最终从红色箭头中流出。

SparkStreaming中提供一些列窗口函数，方便对窗口数据进行分析，文档：

在实际项目中，很多时候需求：每隔一段时间统计最近数据状态，并不是对所有数据进行统计，称为趋势统计或者窗口统计，SparkStreaming中提供相关函数实现功能，业务逻辑如下：
针对用户百度搜索日志数据，实现【近期时间内热搜Top10】，统计最近一段时间范围（比如，最近半个小时或最近2个小时）内用户搜索词次数，获取Top10搜索词及次数。窗口函数【window】声明如下，包含两个参数：窗口大小（WindowInterval，每次统计数据范围）和滑动大小（每隔多久统计一次），都必须是批处理时间间隔BatchInterval整数倍。
案例完整实现代码如下，为了演示方便，假设BatchInterval为2秒，WindowInterval
为4秒，SlideInterval为2秒。

SparkStreaming中同时提供将窗口Window设置与聚合reduceByKey合在一起的函数，为了更加方便编程。
使用【reduceByKeyAndWindow】函数，修改上述代码，实现窗口统计，具体代码如下：

package cn.oldlut.spark.app.window

[外链图片转存中…(img-j2bSqLkn-1715601431319)]
[外链图片转存中…(img-ROwnaN0d-1715601431320)]

网上学习资料一大堆，但如果学到的知识不成体系，遇到问题时只是浅尝辄止，不再深入研究，那么很难做到真正的技术提升。