1. 滚动窗口
– 窗口长度 = 滑动间隔
– 数据不重叠

2. 滑动窗口
– 窗口长度 > 滑动间隔
– 数据有重叠

# 窗口操作示例
scala> val ssc = new StreamingContext(sc, Seconds(5))

# window操作
scala> val windowed = lines.window(Seconds(30), Seconds(10))
windowed: org.apache.spark.streaming.dstream.DStream[String] = WindowedDStream[11]

# reduceByWindow操作
scala> val reducedWindow = lines.reduceByWindow(_ + ” ” + _, Seconds(30), Seconds(10))
reducedWindow: org.apache.spark.streaming.dstream.DStream[String] = ReducedWindowedDStream[12]

# countByWindow操作
scala> val countWindow = lines.countByWindow(Seconds(30), Seconds(10))
countWindow: org.apache.spark.streaming.dstream.DStream[Long] = ForEachDStream[13]

# reduceByKeyAndWindow操作
scala> val wordCountsWindow = wordCounts.reduceByKeyAndWindow(
| (a: Int, b: Int) => a + b, // reduce函数
| (a: Int, b: Int) => a – b, // inverse reduce函数
| Seconds(30), // 窗口长度
| Seconds(10) // 滑动间隔
| )
wordCountsWindow: org.apache.spark.streaming.dstream.DStream[(String, Int)] = ReducedWindowedDStream[14]

scala> wordCountsWindow.print()

# countByValueAndWindow操作
scala> val countByValueWindow = words.countByValueAndWindow(Seconds(30), Seconds(10))
countByValueWindow: org.apache.spark.streaming.dstream.DStream[(String, Long)] = ForEachDStream[15]

# 创建应用
$ cat > /bigdata/spark-apps/FgeduStreamingWordCount.scala << 'EOF' package com.fgedu.spark import org.apache.spark.SparkConf import org.apache.spark.streaming.{Seconds, StreamingContext} import org.apache.spark.streaming.kafka010._ import org.apache.kafka.common.serialization.StringDeserializer object FgeduStreamingWordCount { def main(args: Array[String]): Unit = { // 配置 val conf = new SparkConf() .setAppName("fgedu-streaming-wordcount") .setMaster("spark://192.168.1.60:7077") // 创建StreamingContext val ssc = new StreamingContext(conf, Seconds(5)) ssc.checkpoint("hdfs://192.168.1.60:9000/spark-checkpoint/fgedu-wordcount") // Kafka配置 val kafkaParams = Map[String, Object]( "bootstrap.servers" -> “192.168.1.51:9092”,
“key.deserializer” -> classOf[StringDeserializer],
“value.deserializer” -> classOf[StringDeserializer],
“group.id” -> “fgedu-wordcount-group”,
“auto.offset.reset” -> “latest”,
“enable.auto.commit” -> (false: java.lang.Boolean)
)

// 创建DStream
val topics = Array(“fgedu-streaming-topic”)
val stream = KafkaUtils.createDirectStream[String, String](
ssc,
PreferConsistent,
Subscribe[String, String](topics, kafkaParams)
)

// 处理数据
val lines = stream.map(record => record.value())
val words = lines.flatMap(_.split(” “))
val wordCounts = words.map(word => (word, 1))
.reduceByKeyAndWindow(
(a: Int, b: Int) => a + b,
Seconds(30),
Seconds(10)
)

// 输出结果
wordCounts.print()

// 启动
ssc.start()
ssc.awaitTermination()
}
}
EOF

# 编译并提交应用
$ /bigdata/app/spark/bin/spark-submit \
–class com.fgedu.spark.FgeduStreamingWordCount \
–master spark://192.168.1.60:7077 \
–executor-memory 4g \
–packages org.apache.spark:spark-sql-kafka-0-10_2.12:3.5.1 \
/bigdata/spark-apps/fgedu-streaming-wordcount.jar

# 发送测试数据
$ /bigdata/app/kafka/bin/kafka-console-producer.sh \
–topic fgedu-streaming-topic \
–bootstrap-server 192.168.1.51:9092

>hello world hello spark
>hello kafka hello fgedu
>spark streaming spark real-time

# 查看输出结果
——————————————-
Time: 2026-04-08 120000
——————————————-
(hello,4)
(spark,3)
(world,1)
(kafka,1)
(fgedu,1)
(streaming,1)
(real-time,1)

scala> val ssc = new StreamingContext(sc, Seconds(5))
scala> ssc.checkpoint(“hdfs://192.168.1.60:9000/spark-checkpoint/fgedu-join”)

# 创建静态数据（维度表）
scala> val dimData = Seq(
| (“IT”, “信息技术部”),
| (“HR”, “人力资源部”),
| (“Finance”, “财务部”)
| )
dimData: Seq[(String, String)] = List((IT,信息技术部), (HR,人力资源部), (Finance,财务部))

scala> val dimDF = dimData.toDF(“dept_code”, “dept_name”)
dimDF: org.apache.spark.sql.DataFrame = [dept_code: string, dept_name: string]

# 广播维度数据
scala> val dimBroadcast = sc.broadcast(dimDF.collect())
dimBroadcast: org.apache.spark.broadcast.Broadcast[Array[org.apache.spark.sql.Row]] = Broadcast(0)

# 创建流数据
scala> val streamData = KafkaUtils.createDirectStream[String, String](
| ssc,
| PreferConsistent,
| Subscribe[String, String](Array(“fgedu-streaming-topic”), kafkaParams)
| )

# 解析流数据
scala> import org.apache.spark.sql.Row
scala> val parsedStream = streamData.map(record => {
| val parts = record.value().split(“,”)
| (parts(0), parts(1), parts(2)) // (name, dept_code, salary)
| })

# 流批Join
scala> val enrichedStream = parsedStream.map(record => {
| val dimMap = dimBroadcast.value.map(r => (r.getString(0), r.getString(1))).toMap
| val deptName = dimMap.getOrElse(record._2, “Unknown”)
| (record._1, deptName, record._3)
| })

scala> enrichedStream.print()

# 排查步骤
# 1. 检查处理时间
# 在Spark UI的Streaming Tab查看Processing Time

# 2. 检查资源使用
$ curl http://192.168.1.60:8080/json

# 解决方案
# 1. 增加批处理间隔
val ssc = new StreamingContext(sc, Seconds(10))

# 2. 增加并行度
spark.default.parallelism=200

# 3. 启用背压
spark.streaming.backpressure.enabled=true
spark.streaming.backpressure.initialRate=1000

# 4. 增加Executor资源
–executor-memory 8g
–num-executors 10

# 排查步骤
# 1. 检查检查点配置
scala> ssc.checkpoint(“hdfs://192.168.1.60:9000/spark-checkpoint”)

# 2. 检查Kafka offset提交

# 解决方案
# 1. 启用Write Ahead Logs
spark.streaming.receiver.writeAheadLog.enable=true

# 2. 配置精确一次语义
spark.streaming.kafka.maxRatePerPartition=1000

# 3. 手动提交offset
stream.foreachRDD { rdd =>
val offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
// 处理数据
// 提交offset
stream.asInstanceOf[CanCommitOffsets].commitAsync(offsetRanges)
}

# 部署最佳实践
1. 配置资源充足
2. 启用WAL提高可靠性
3. 监控应用状态
4. 配置告警机制
5. 定期清理检查点