Spark Streaming核心编程总结（四）

 一、有状态转化操作：UpdateStateByKey
概念与作用
UpdateStateByKey 用于在流式计算中跨批次维护状态（如累加统计词频）。它允许基于键值对形式的DStream，通过自定义状态更新函数，将历史状态与新数据结合，生成包含最新状态的DStream。

实现步骤
1. 定义状态类型：状态可以是任意数据类型（如示例中的Int类型）。
2. 定义状态更新函数：接收当前批次的数据序列和旧状态，返回新状态。
 

val updateFunc = (values: Seq[Int], state: Option[Int]) => {
val currentCount = values.sum
val previousCount = state.getOrElse(0)
Some(currentCount + previousCount)
}

3. 配置检查点目录：必须设置检查点以持久化状态，确保容错性。

ssc.checkpoint("./ck")

4. 应用操作：通过updateStateByKey将函数作用于键值对DStream。

val stateDStream = pairs.updateStateByKey[Int](updateFunc)

 二、窗口操作：WindowOperations
概念与作用
窗口操作基于时间窗口动态处理数据，适用于滑动统计（如最近12秒内的词频）。需定义两个参数：
 窗口时长：计算的时间范围（如Seconds(12)）。
 滑动步长：触发计算的间隔（如Seconds(6)）。

实现示例

val wordCounts = pairs.reduceByKeyAndWindow(
(a: Int, b: Int) => a + b, // 聚合函数
Seconds(12), // 窗口时长
Seconds(6) // 滑动步长
)

 三、DStream输出操作
输出操作触发DStream的实际计算，支持多种数据落地方式：

1. 基础输出
 print()：打印每批次前10个元素，用于调试。
 saveAsTextFiles / saveAsObjectFiles / saveAsHadoopFiles：将数据保存为文本、序列化文件或Hadoop格式。

2. 通用输出：foreachRDD
允许对每个RDD执行自定义操作（如写入数据库）。需注意：
 连接管理：避免在Driver端创建连接（序列化问题），应在foreachPartition中按分区创建。
 资源优化：每个分区建立一次连接，而非每条数据，减少开销。

示例：

wordCounts.foreachRDD { rdd =>
rdd.foreachPartition { partition =>
val connection = createDatabaseConnection()
partition.foreach(data => connection.write(data))
connection.close()
}
}