img
img
img

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

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

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

  • 方法二：consumerKafka，消费Kafka Topic中数据在这里插入图片描述
    具体代码如下：
package cn.oldlut.spark.app

import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.DStream
import org.apache.spark.streaming.kafka010._
import org.apache.spark.streaming.{Seconds, StreamingContext}

/\*\*
 \* 工具类提供：构建流式应用上下文StreamingContext实例对象和从Kafka Topic消费数据
 \*/
object StreamingContextUtils {
  /\*\*
 \* 获取StreamingContext实例，传递批处理时间间隔
 \*
 \* @param batchInterval 批处理时间间隔，单位为秒
 \*/
  def getStreamingContext(clazz: Class[_], batchInterval: Int): StreamingContext = {
    // i. 创建SparkConf对象，设置应用配置信息
    val sparkConf = new SparkConf()
      .setAppName(clazz.getSimpleName.stripSuffix("$"))
      .setMaster("local[3]")
      // 设置Kryo序列化
      .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
      .registerKryoClasses(Array(classOf[ConsumerRecord[String, String]]))
    // ii.创建流式上下文对象, 传递SparkConf对象和时间间隔
    val context = new StreamingContext(sparkConf, Seconds(batchInterval))
    // iii. 返回
    context
  }

  /\*\*
 \* 从指定的Kafka Topic中消费数据，默认从最新偏移量（largest）开始消费
 \*
 \* @param ssc StreamingContext实例对象
 \* @param topicName 消费Kafka中Topic名称
 \*/
  def consumerKafka(ssc: StreamingContext, topicName: String): DStream[ConsumerRecord[String, String]] = {
    // i.位置策略
    val locationStrategy: LocationStrategy = LocationStrategies.PreferConsistent
    // ii.读取哪些Topic数据
    val topics = Array(topicName)
    // iii.消费Kafka 数据配置参数
    val kafkaParams = Map[String, Object](
      "bootstrap.servers" -> "node1.oldlut.cn:9092",
      "key.deserializer" -> classOf[StringDeserializer],
      "value.deserializer" -> classOf[StringDeserializer],
      "group.id" -> "group\_id\_streaming\_0001",
      "auto.offset.reset" -> "latest",
      "enable.auto.commit" -> (false: java.lang.Boolean)
    )
    // iv.消费数据策略
    val consumerStrategy: ConsumerStrategy[String, String] = ConsumerStrategies.Subscribe(
      topics, kafkaParams
    )
    // v.采用新消费者API获取数据，类似于Direct方式
    val kafkaDStream: DStream[ConsumerRecord[String, String]] = KafkaUtils.createDirectStream(
      ssc, locationStrategy, consumerStrategy
    )
    // vi.返回DStream
    kafkaDStream
  }
}

3 实时数据ETL存储

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

package cn.oldlut.spark.app.etl

import cn.oldlut.spark.app.StreamingContextUtils
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.StreamingContext
import org.apache.spark.streaming.dstream.DStream
import org.lionsoul.ip2region.{DataBlock, DbConfig, DbSearcher}

/\*\*
 \* 实时消费Kafka Topic数据，经过ETL（过滤、转换）后，保存至HDFS文件系统中，BatchInterval为：10s
 \*/
object StreamingETLHdfs {
  def main(args: Array[String]): Unit = {
    // 1. 获取StreamingContext实例对象
    val ssc: StreamingContext = StreamingContextUtils.getStreamingContext(this.getClass, 10)
    // 2. 从Kafka消费数据，使用Kafka New Consumer API
    val kafkaDStream: DStream[ConsumerRecord[String, String]] = StreamingContextUtils
      .consumerKafka(ssc, "search-log-topic")
    // 3. 数据ETL：过滤不合格数据及转换IP地址为省份和城市，并存储HDFS上
    kafkaDStream.foreachRDD { (rdd, time) =>
      // i. message不为null，且分割为4个字段
      val kafkaRDD: RDD[ConsumerRecord[String, String]] = rdd.filter { record =>
        val message: String = record.value()
        null != message && message.trim.split(",").length == 4
      }
      // ii. 解析IP地址
      val etlRDD: RDD[String] = kafkaRDD.mapPartitions { iter =>
        // 创建DbSearcher对象，针对每个分区创建一个，并不是每条数据创建一个
        val dbSearcher = new DbSearcher(new DbConfig(), "dataset/ip2region.db")
        iter.map { record =>
          val Array(_, ip, _, _) = record.value().split(",")
          // 依据IP地址解析
          val dataBlock: DataBlock = dbSearcher.btreeSearch(ip)
          val region: String = dataBlock.getRegion
          val Array(_, _, province, city, _) = region.split("\\|")
          // 组合字符串
          s"${record.value()},$province,$city"
        }
      }
      // iii. 保存至文件
      val savePath = s"datas/streaming/etl/search-log-${time.milliseconds}"
      if (!etlRDD.isEmpty()) {
        etlRDD.coalesce(1).saveAsTextFile(savePath)
      }
    }
    // 4.启动流式应用，一直运行，直到程序手动关闭或异常终止
    ssc.start()
    ssc.awaitTermination()
    ssc.stop(stopSparkContext = true, stopGracefully = true)
  }
}

运行模拟日志数据程序和ETL应用程序，查看实时数据ETL后保存文件，截图如下：在这里插入图片描述

4 实时状态更新统计

实 时 累 加 统 计 用 户 各 个 搜 索 词 出 现 的 次 数 ， 在 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检查点目录：在这里插入图片描述
完整演示代码如下：

package cn.oldlut.spark.app.state

import cn.oldlut.spark.app.StreamingContextUtils
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.spark.streaming.StreamingContext
import org.apache.spark.streaming.dstream.DStream

/\*\*
 \* 实时消费Kafka Topic数据，累加统计各个搜索词的搜索次数，实现百度搜索风云榜
 \*/
object StreamingUpdateState {
  def main(args: Array[String]): Unit = {
    // 1. 获取StreamingContext实例对象
    val ssc: StreamingContext = StreamingContextUtils.getStreamingContext(this.getClass, 5)
    // TODO: 设置检查点目录
    ssc.checkpoint(s"datas/streaming/state-${System.nanoTime()}")
    // 2. 从Kafka消费数据，使用Kafka New Consumer API
    val kafkaDStream: DStream[ConsumerRecord[String, String]] = StreamingContextUtils
      .consumerKafka(ssc, "search-log-topic")
    // 3. 对每批次的数据进行搜索词次数统计
    val reduceDStream: DStream[(String, Int)] = kafkaDStream.transform { rdd =>
      val reduceRDD = rdd
        // 过滤不合格的数据
        .filter { record =>
          val message: String = record.value()
          null != message && message.trim.split(",").length == 4
        }
        // 提取搜索词，转换数据为二元组，表示每个搜索词出现一次
        .map { record =>
          val keyword: String = record.value().trim.split(",").last
          keyword -> 1
        }
        // 按照单词分组，聚合统计
        .reduceByKey((tmp, item) => tmp + item) // TODO: 先聚合，再更新，优化
      reduceRDD // 返回
    }
    /\*
 def updateStateByKey[S: ClassTag](
 // 状态更新函数
 updateFunc: (Seq[V], Option[S]) => Option[S]
 ): DStream[(K, S)]
 第一个参数：Seq[V]
 表示的是相同Key的所有Value值
 第二个参数：Option[S]
 表示的是Key的以前状态，可能有值Some，可能没值None，使用Option封装
 S泛型，具体类型有业务具体，此处是词频：Int类型
 \*/
    val stateDStream: DStream[(String, Int)] = reduceDStream.updateStateByKey(
      (values: Seq[Int], state: Option[Int]) => {
        // a. 获取以前状态信息
        val previousState = state.getOrElse(0)
        // b. 获取当前批次中Key对应状态
        val currentState = values.sum
        // c. 合并状态
        val latestState = previousState + currentState
        // d. 返回最新状态
        Some(latestState)
      }
    )
    // 5. 将结果数据输出 -> 将每批次的数据处理以后输出
    stateDStream.print()
    // 6.启动流式应用，一直运行，直到程序手动关闭或异常终止
    ssc.start()
    ssc.awaitTermination()
    ssc.stop(stopSparkContext = true, stopGracefully = true)
  }
}

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

4.2 mapWithState 函数

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

package cn.oldlut.spark.app.state

import cn.oldlut.spark.app.StreamingContextUtils
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.{State, StateSpec, StreamingContext}
import org.apache.spark.streaming.dstream.DStream

/\*\*
 \* 实时消费Kafka Topic数据，累加统计各个搜索词的搜索次数，实现百度搜索风云榜
 \*/
object StreamingMapWithState {
  def main(args: Array[String]): Unit = {
    // 1. 获取StreamingContext实例对象
    val ssc: StreamingContext = StreamingContextUtils.getStreamingContext(this.getClass, 5)
    // TODO: 设置检查点目录
    ssc.checkpoint(s"datas/streaming/state-${System.nanoTime()}")
    // 2. 从Kafka消费数据，使用Kafka New Consumer API
    val kafkaDStream: DStream[ConsumerRecord[String, String]] = StreamingContextUtils
      .consumerKafka(ssc, "search-log-topic")
    // 3. 对每批次的数据进行搜索词进行次数统计
    val reduceDStream: DStream[(String, Int)] = kafkaDStream.transform { rdd =>
      val reduceRDD: RDD[(String, Int)] = rdd
        // 过滤不合格的数据
        .filter { record =>
          val message: String = record.value()
          null != message && message.trim.split(",").length == 4
        }
        // 提取搜索词，转换数据为二元组，表示每个搜索词出现一次
        .map { record =>
          val keyword: String = record.value().trim.split(",").last
          keyword -> 1
        }
        // 按照单词分组，聚合统计
        .reduceByKey((tmp, item) => tmp + item) // TODO: 先聚合，再更新，优化
      // 返回
      reduceRDD
    }
    // TODO: 4、实时累加统计搜索词搜索次数，使用mapWithState函数
    /\*
 按照Key来更新状态的，一条一条数据的更新状态
 def mapWithState[StateType: ClassTag, MappedType: ClassTag](
 spec: StateSpec[K, V, StateType, MappedType]
 ): MapWithStateDStream[K, V, StateType, MappedType]
 a. 通过函数源码发现参数使用对象
 StateSpec 实例对象
 b. StateSpec
 表示对状态封装，里面涉及到相关数据类型
 c. 如何构建StateSpec对象实例呢？？
 StateSpec 伴生对象中function函数构建对象
 def function[KeyType, ValueType, StateType, MappedType](
 // 从函数名称可知，针对每条数据更新Key的转态信息
 mappingFunction: (KeyType, Option[ValueType], State[StateType]) => MappedType
 ): StateSpec[KeyType, ValueType, StateType, MappedType]
 \*/
    // 状态更新函数，针对每条数据进行更新状态
    val spec: StateSpec[String, Int, Int, (String, Int)] = StateSpec.function(
      // (KeyType, Option[ValueType], State[StateType]) => MappedType
      (keyword: String, countOption: Option[Int], state: State[Int]) => {
        // a. 获取当前批次中搜索词搜索次数
        val currentState: Int = countOption.getOrElse(0)
        // b. 从以前状态中获取搜索词搜索次数
        val previousState = state.getOption().getOrElse(0)
        // c. 搜索词总的搜索次数
        val latestState = currentState + previousState
        // d. 更行状态
        state.update(latestState)
        // e. 返回最新省份销售订单额
        (keyword, latestState)
      }
    )
    // 调用mapWithState函数进行实时累加状态统计
    val stateDStream: DStream[(String, Int)] = reduceDStream.mapWithState(spec)
    // 5. 将结果数据输出 -> 将每批次的数据处理以后输出
    stateDStream.print()
    // 6.启动流式应用，一直运行，直到程序手动关闭或异常终止
    ssc.start()
    ssc.awaitTermination()
    ssc.stop(stopSparkContext = true, stopGracefully = true)
  }
}

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

5 实时窗口统计

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

http://spark.apache.org/docs/2.4.5/streaming-programming-guide.html#window-operations

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

package cn.oldlut.spark.app.window

import cn.oldlut.spark.app.StreamingContextUtils
import org.apache.spark.streaming.dstream.DStream
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
 * 实时消费Kafka Topic数据，每隔一段时间统计最近搜索日志中搜索词次数
 * 批处理时间间隔：BatchInterval = 2s
 * 窗口大小间隔：WindowInterval = 4s
 * 滑动大小间隔：SliderInterval = 2s
 */
object StreamingWindow {
  def main(args: Array[String]): Unit = {
    // Streaming应用BatchInterval
    val BATCH_INTERVAL: Int = 2
    // Streaming应用窗口大小
    val WINDOW_INTERVAL: Int = BATCH_INTERVAL * 2
    val SLIDER_INTERVAL: Int = BATCH_INTERVAL * 1
    // 1. 获取StreamingContext实例对象
    val ssc: StreamingContext = StreamingContextUtils.getStreamingContext(this.getClass, BATCH_INTERVAL)
    // 2. 从Kafka消费数据，使用Kafka New Consumer API
    val kafkaDStream: DStream[String] = StreamingContextUtils
      .consumerKafka(ssc, "search-log-topic")
      .map(record => record.value())
    // TODO: 添加窗口，设置对应参数
    /*
    def window(windowDuration: Duration, slideDuration: Duration): DStream[T]
    警告信息：
    ERROR KafkaRDD: Kafka ConsumerRecord is not serializable.
    Use .map to extract fields before calling .persist or .window
    */
    val windowDStream: DStream[String] = kafkaDStream.window(
      Seconds(WINDOW_INTERVAL), Seconds(SLIDER_INTERVAL)
    )
    // 4. 对每批次的数据进行搜索词进行次数统计
    val countDStream: DStream[(String, Int)] = windowDStream.transform { rdd =>
      val resultRDD = rdd
        // 过滤不合格的数据
        .filter(message => null != message && message.trim.split(",").length == 4)
        // 提取搜索词，转换数据为二元组，表示每个搜索词出现一次
        .map { message =>
          val keyword: String = message.trim.split(",").last
          keyword -> 1
        }
        // 按照单词分组，聚合统计
        .reduceByKey((tmp, item) => tmp + item)
      // 返回
      resultRDD
    }
    // 5. 将结果数据输出 -> 将每批次的数据处理以后输出
    countDStream.print()
    // 6.启动流式应用，一直运行，直到程序手动关闭或异常终止
    ssc.start()
    ssc.awaitTermination()
    ssc.stop(stopSparkContext = true, stopGracefully = true)
  }
}

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

package cn.oldlut.spark.app.window


![img](https://img-blog.csdnimg.cn/img_convert/bbc4de80fab3c3f0e93e759c13344271.png)
![img](https://img-blog.csdnimg.cn/img_convert/891b2231b61eeb1789e930ad89425e98.png)

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

**[需要这份系统化资料的朋友，可以戳这里获取](https://bbs.csdn.net/topics/618545628)**


**一个人可以走的很快，但一群人才能走的更远！不论你是正从事IT行业的老鸟或是对IT行业感兴趣的新人，都欢迎加入我们的的圈子（技术交流、学习资源、职场吐槽、大厂内推、面试辅导），让我们一起学习成长！**

F,t_70,g_se,x_16)  
 使用【reduceByKeyAndWindow】函数，修改上述代码，实现窗口统计，具体代码如下：

package cn.oldlut.spark.app.window

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

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

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

一个人可以走的很快，但一群人才能走的更远！不论你是正从事IT行业的老鸟或是对IT行业感兴趣的新人，都欢迎加入我们的的圈子（技术交流、学习资源、职场吐槽、大厂内推、面试辅导），让我们一起学习成长！

# 大数据 # 面试 # 学习
Logo
技术共进，成长同行——讯飞AI开发者社区

技术共进，成长同行——讯飞AI开发者社区

更多推荐

AI赋能：智能电网的未来革命

智能电网通过传感器、智能电表和物联网设备收集海量数据，包括用电量、电压、电流、频率等实时信息。这些数据为人工智能提供了丰富的输入源，使其能够进行预测、优化和自动化决策。人工智能技术在智能电网中的应用主要体现在负荷预测、故障检测、能源调度和分布式能源管理等方面。结合图神经网络，可以分析电网拓扑结构中的故障传播路径，快速定位问题源头。能源调度是智能电网的核心问题，需要在满足需求的同时最小化成本和碳排放

avatar 讯飞AI开发者社区

AI赋能智能家居：30字揭秘能耗管理黑科技

随着算法进步和硬件发展，未来的智能家居系统将实现更高精度的能耗管理和更自然的用户交互。人工智能模型可以识别能耗 patterns，预测未来能耗需求，并自动调整设备运行参数。深度学习算法在处理非结构化数据方面表现出色，能够从复杂的能耗数据中提取有价值的信息。智能家居设备产生的大数据为能耗管理提供了丰富的信息源。环境模拟智能家居系统的状态变化，智能体通过不断尝试和学习，找到最小化能耗的最优策略。这种方

avatar 讯飞AI开发者社区

联邦学习：隐私保护的AI新范式

联邦学习（Federated Learning）是一种分布式机器学习框架，旨在保护数据隐私的同时利用分散的数据进行模型训练。联邦学习技术仍在快速发展中，几个重要方向包括：更高效的隐私保护机制、跨模态联邦学习、联邦学习与区块链的结合，以及标准化框架的建立。这些进步将进一步推动联邦学习在医疗、金融和智能物联网等领域的应用。通过上述方法和示例代码可以看出，联邦学习为隐私保护下的分布式数据利用提供了可行的

avatar 讯飞AI开发者社区