做实时数据处理这几年，Spark Streaming基本是我接手新项目时第一个会碰的组件。不管你是做日志实时分析、用户行为埋点、指标监控，还是简单的消息流转，只要你用的是Spark生态，那Spark Streaming就是绕不开的一环。尤其到了Spark 3.x时代，Structured Streaming已经非常成熟，很多人还在写老的DStream API，这就有点跟不上节奏了。这篇文章我打算把手里的实操经验和踩过的坑都翻出来，从核心概念讲到生产环境配置，顺便把DStream和Structured Streaming这两套API的路线差异说清楚，给还没入门的同学一条比较顺的上手路径。

作为一个系列的第4篇，这篇不会去重复Spark RDD、DataFrame这些基础，默认你已经能跑通一个简单的Spark作业。文章会专注于“流”这件事：数据怎么进来、怎么切分成微批、状态怎么管理、结果怎么输出，以及在生产上最容易翻车的那几个细节。

1. 先把Spark Streaming看明白

1.1 它到底做什么，解决什么问题

Spark Streaming本质上是一个“微批处理”引擎。它把连续不断的实时数据流，按固定时间间隔切成一批一批的小数据，然后交给Spark的批处理引擎去算。所以你表面上写的是流处理代码，底层跑的其实还是RDD或DataFrame上的批量任务。

这么设计的好处很明显：第一，它完全复用Spark批处理那套成熟的容错机制，任务失败了可以重新调度，血统一套链上去，爽快；第二，它不需要单独维护一套流计算引擎的集群，一个Spark集群既能跑批又能跑流，运维成本低。第三，它天然就能和Spark生态里的SQL、MLlib结合起来，做流批一体很方便。

那它解决的是什么问题？典型场景包括：

• 日志采集与实时分析：比如你有一个Nginx日志流，想实时统计UV、PV、错误率。
• 用户行为实时监控：App埋点事件进来后，马上算出当前的活跃用户、核心漏斗转化率。
• 指标告警与监控：对机器指标做滑动窗口聚合，阈值触发了立刻告警。
• 实时数仓的ODS层：把Kafka里的事件流实时写入数据仓库的明细层，供后续分析。

一句话总结：如果你的业务对延迟的要求是秒级到分钟级，而且你已经有一支熟悉Spark的团队，那Spark Streaming是一个非常务实的选择。

1.2 Batch Interval到底怎么定

微批处理里有几个核心参数，最先要理解的就是批处理间隔（batch interval）。它决定了下游任务的最小延迟和计算粒度。你设成5秒，那每5秒就会触发一次批处理，数据在这5秒内累积，然后一次性处理完。

选这个间隔有讲究。设小了，实时性高，但任务调度开销会变大，处理不过来时就会出现“批处理积压”；设大了，吞吐量上去了，但延迟也变高了。我一般会先拍一个值，然后用压力测试去调。

实际操作中，我的建议是这样：

• 日志采集类，5~10秒比较常见。
• 用户行为分析，如果业务方接受20秒内的延迟，可以设到10秒。
• 某些对延迟要求高的监控场景，可以考虑1~2秒，但你要做好资源冗余的准备。

你可以在代码里这么设：

val ssc = new StreamingContext(sparkConf, Seconds(5))

这个5秒就是你要不断去调整的核心参数。要注意的是，它不只是影响延迟，还影响你每个批次的记录数、内存占用、shuffle量。批处理间隔设得越大，每个批次处理的数据越多，对executor内存的压力就越大。

2. 核心概念拆解：DStream、窗口与状态

2.1 DStream是“老的”还是“还能用”

先承认一个现状：官方对DStream API的态度是维护为主，不再投入大量新功能开发。但网上大量教程和开源项目还是基于DStream写的，所以你得能看懂它。

DStream的本质是“RDD的序列”，也就是一系列连续的RDD。每个RDD对应一个批次的数据。你基于DStream做map、flatMap、reduceByKey这些操作时，其实是在对序列里的每个RDD做转换。

举一个最简单的例子，从socket读数据做词频统计：

val sparkConf = new SparkConf().setAppName("NetworkWordCount") val ssc = new StreamingContext(sparkConf, Seconds(5)) val lines = ssc.socketTextStream("localhost", 9999) val words = lines.flatMap(_.split(" ")) val pairs = words.map(word => (word, 1)) val wordCounts = pairs.reduceByKey(_ + _) wordCounts.print() ssc.start() ssc.awaitTermination()

这段代码你拿去一把梭，是可以跑的。它每5秒统计一次当前socket收到的那批文本里各单词出现了几次。这个结果是无状态的，下一批数据来了，上一批的计数就没了。

DStream API有一个很明显的痛点：它基于RDD编程，意味着你要习惯函数式编程，同时调试起来比较费劲，因为DStream的转换逻辑你看不到中间结果。而且它对SQL、DataFrame的支持很生硬，做结构化数据处理时特别别扭。所以到了Spark 2.x之后，官方推出了Structured Streaming，我自己的新项目基本都用Structured Streaming了。

2.2 窗口操作与滑动窗口

窗口是流处理里的重头戏，它让你能在“一段时间”上做聚合。比如你想统计过去1分钟内某个接口的调用量，每10秒刷新一次，就需要窗口操作。窗口有两个参数：窗口长度（window length）和滑动间隔（slide interval）。

看图比看字容易理解：窗口长度决定了你聚合的数据覆盖多长时间范围，滑动间隔决定了你多长时间重新计算一次。如果窗口长度等于滑动间隔，那就是一个滚动窗口（tumbling window），不会重叠；如果滑动间隔小于窗口长度，就是滑动窗口，会有数据重叠，而且同一个数据可能出现在多个窗口里。

DStream里窗口操作的写法是这样：

val windowedCounts = pairs.reduceByKeyAndWindow( (a: Int, b: Int) => a + b, // 窗口内的聚合函数 Seconds(30), // 窗口长度30秒 Seconds(10) // 每10秒滑动一次 )

这段代码的逻辑是：每10秒钟计算一次最近30秒的单词计数。窗口里的数据会有一部分和上一个窗口重叠，Spark会在内部做一些优化，但你不用管这些。

Structured Streaming里的窗口写法更清晰，它是基于事件时间的：

val windowedCounts = events .groupBy(window($"eventTime", "5 minutes"), $"word") .count()

注意这里有个巨大的差别：DStream的窗口处理的是“到达时间”，也就是数据进入系统的时刻；Structured Streaming的窗口可以基于数据里自带的事件时间（event time），精确地还原业务时刻。这一点对于处理乱序数据非常重要。

2.3 状态管理与updateStateByKey

很多场景不满足于无状态的计算。比如你想统计从启动到现在每个用户的累计访问次数，或者某个传感器从上线以来的累计温度均值。这就需要有状态的计算。

DStream里做状态计算，最常用的一个API是updateStateByKey。它允许你在每个批次结束时，把当前批次的数据和之前保存下来的状态做一个合并，然后输出新状态。

一个典型的计数器示例：

def updateFunction(newValues: Seq[Int], runningCount: Option[Int]): Option[Int] = { val newCount = runningCount.getOrElse(0) + newValues.sum Some(newCount) } val runningCounts = pairs.updateStateByKey[Int](updateFunction _)

这个API用起来有几个坑要注意。第一个坑是：状态数据是保存在checkpoint里的，一旦checkpoint目录失联或者数据损坏，状态就丢了。第二个坑是：state数据默认存储在内存中，如果state的总量超过了executor内存，会直接OOM。所以状态特别大的场景，你得设计方案提前处理，比如按key做哈希分区，或者定期清理冷数据。

到了Structured Streaming里，状态管理被封装到了Streaming Aggregation里，你写groupBy即可，Spark自动帮你管理状态。状态后端也换成了RocksDB，这对超大状态场景友好得多。如果你用的是Spark 3.x，强烈建议把状态后端切到RocksDB。

3. 实操：DStream API 写一个Kafka消费程序

3.1 环境与依赖准备

在生产环境里，DStream最常见的数据源就是Kafka。假设你已经有了一套Kafka在跑，topic名称是“log-input”，里面每条消息是JSON格式的访问日志。我们准备写一个Spark Streaming作业，消费这个topic，统计每秒钟的请求次数和平均耗时。

先准备依赖。我的项目是Maven管理的，核心依赖长这样：

<dependency> <groupId>org.apache.spark</groupId> <artifactId>spark-streaming_2.12</artifactId> <version>3.3.2</version> </dependency> <dependency> <groupId>org.apache.spark</groupId> <artifactId>spark-streaming-kafka-0-10_2.12</artifactId> <version>3.3.2</version> </dependency>

注意spark-streaming-kafka-0-10这个依赖的版本号中有个0-10，它对应的是Kafka API的版本，不是Spark版本。在Spark 3.x里，这个依赖已经内置了对Kafka 2.x版本的适配，你不需要再单独引入Kafka的客户端jar包。

3.2 消费逻辑与代码实现

创建StreamingContext和DirectStream的完整代码这样写：

import org.apache.kafka.clients.consumer.ConsumerRecord import org.apache.kafka.common.serialization.StringDeserializer import org.apache.spark.SparkConf import org.apache.spark.streaming.{Seconds, StreamingContext} import org.apache.spark.streaming.kafka010._ val sparkConf = new SparkConf().setAppName("KafkaLogAnalyzer") val ssc = new StreamingContext(sparkConf, Seconds(5)) ssc.checkpoint("hdfs:///tmp/spark-streaming-cp") val kafkaParams = Map[String, Object]( "bootstrap.servers" -> "kafka01:9092,kafka02:9092", "key.deserializer" -> classOf[StringDeserializer], "value.deserializer" -> classOf[StringDeserializer], "group.id" -> "spark-streaming-log-analyzer", "auto.offset.reset" -> "latest", "enable.auto.commit" -> (false: java.lang.Boolean) ) val topics = Array("log-input") val stream = KafkaUtils.createDirectStream[String, String]( ssc, LocationStrategies.PreferConsistent, ConsumerStrategies.Subscribe[String, String](topics, kafkaParams) )

这里有几个点值得说一下：

• enable.auto.commit必须设置成false，就是为了不让Kafka的offset自动提交和Spark的处理进度脱节。
• auto.offset.reset设置成latest还是earliest，取决于你的业务。如果是实时数据分析，用latest；如果你是补数或重新处理历史数据，用earliest。
• LocationStrategies.PreferConsistent是一种分布策略，它会让Spark的executor尽量和Kafka partition的leader保持一致，减少网络传输开销。

拿到DStream之后，我们对每条记录做解析：

import org.json4s._ import org.json4s.jackson.JsonMethods._ implicit val formats = DefaultFormats case class LogRecord(timestamp: Long, api: String, costMs: Long, status: Int) val parsed = stream.map(_.value()).map { json => try { val obj = parse(json) val record = obj.extract[LogRecord] Some(record) } catch { case _: Exception => None } }.filter(_.isDefined).map(_.get) val requestCount = parsed.map(r => (r.api, 1L)) val totalCost = parsed.map(r => (r.api, r.costMs)) val stats = requestCount.union(totalCost) .reduceByKeyAndWindow( (a: (Long, Long), b: (Long, Long)) => (a._1 + b._1, a._2 + b._2), (a: (Long, Long), b: (Long, Long)) => (a._1 - b._1, a._2 - b._2), Seconds(60), Seconds(10) ) stats.foreachRDD { rdd => val result = rdd.collect() result.foreach { case (api, (count, cost)) => val avgCost = if (count == 0) 0.0 else cost.toDouble / count.toDouble println(s"API=$api count=$count avgCost=${avgCost}ms") } } ssc.start() ssc.awaitTermination()

你会注意到我这里reduceByKeyAndWindow多传了一个反向聚合函数。这个反向函数的作用是让Spark在窗口滑动时，可以增量地“减掉”移出窗口的老数据，而不是每个窗口都全量重算一遍。这个性能收益在窗口很大、数据量很大的时候特别明显。

3.3 运行与验证

在集群上提交作业：

spark-submit \ --class com.example.KafkaLogAnalyzer \ --master yarn \ --deploy-mode cluster \ --executor-memory 4G \ --num-executors 4 \ --executor-cores 2 \ spark-streaming-demo-1.0.0.jar

跑起来之后，你可以观察几个指标：

• 批处理时间（Batch Processing Time）：每个批次从开始到结束花费了多久。
• 调度延迟（Scheduling Delay）：批次在队列里等了多久才被处理。
• 总延迟（Total Delay）：从数据产生到被处理的端到端延迟。

如果发现批处理时间经常大于批处理间隔，说明你的作业在积压，要么加资源，要么优化逻辑，要么调大批处理间隔。这里最直观的判断方法是看Spark UI里的Streaming标签页，它有一个quarterly的状态汇总，能直接看到每个批次的处理耗时曲线。

4. Structured Streaming 实操：从入门到生产

4.1 重新设计一个流处理任务

Structured Streaming用起来和DStream完全是两种体验。它最大的特点是：你把流数据当成一张“无限追加的表”，用Spark SQL那一套DataFrame算子去处理它，输出结果可以是一个新的流。

我用Structured Streaming重写刚才的日志统计需求：

import org.apache.spark.sql.SparkSession import org.apache.spark.sql.functions._ val spark = SparkSession.builder() .appName("StructuredLogAnalyzer") .getOrCreate() import spark.implicits._ val rawDF = spark.readStream .format("kafka") .option("kafka.bootstrap.servers", "kafka01:9092,kafka02:9092") .option("subscribe", "log-input") .option("startingOffsets", "latest") .load() val logs = rawDF.selectExpr("CAST(value AS STRING) as json") .select( get_json_object($"json", "$.timestamp").cast("long").as("ts"), get_json_object($"json", "$.api").as("api"), get_json_object($"json", "$.costMs").cast("long").as("costMs") ) val stats = logs .withWatermark("ts", "30 seconds") .groupBy(window($"ts", "5 minutes"), $"api") .agg( count("*").as("request_count"), avg("costMs").as("avg_cost") ) val query = stats.writeStream .outputMode("append") .format("console") .option("truncate", "false") .start() query.awaitTermination()

这段代码能跑通的一个前提是：你的Kafka消息里的timestamp字段是合法的Long类型，表示Unix时间戳（毫秒）。如果字段类型不对，后面的窗口聚合会直接报错或结果为空。

这里我特别加了一个withWatermark操作，后面的4.2小节我会仔细讲它为什么要存在。

4.2 事件时间与水印：乱序数据的救星

Kafka里的数据不一定是按产生顺序到达的。上游网络抖动、应用程序重试、消息积压后的大批涌入，都会导致乱序。如果你还按照“数据到达时间”做窗口计算，统计出来的结果就是不准确的。

Structured Streaming引入了事件时间概念，它允许你决定“用哪个字段当作业务时间戳”，并且通过watermark机制来处理迟到的数据。

watermark的意思是：你认为在一个事件时间点之后，超过指定阈值的数据就不值得再等了。比如你设置watermark为“10分钟”，那么当处理引擎看到事件时间12:30的数据时，它会认为事件时间早于12:20的数据已经不会再来，或者就算来了也不处理了。这样，系统就可以安全地关闭12:20之前的窗口状态，释放内存。

代码里的含义是：

.withWatermark("ts", "30 seconds")

这表示允许最多30秒的乱序或延迟。如果你的上游Kafka到Spark之间的链路比较稳，30秒够用；一旦网络抖动或者Spark重启恢复期间Kafka积压了大量数据，这个值就要调大。不过也别无脑调大，watermark越大，窗口状态保留的时间越长，内存和RocksDB的压力也会更大。

生产上我的经验是：先小流量试跑3天，用真实数据里的最大延迟来定watermark，不要拍脑袋。

4.3 输出模式与Sink选择

Structured Streaming有3种输出模式：

• append：只输出“新增”的行。窗口聚合里一般配合水印使用，当窗口关闭后才输出该窗口的结果。
• complete：每次更新时输出全量的结果表。适合结果集不大的场景，比如全局的UV统计。
• update：每次更新时输出变化过的行。这个比较灵活，适合结果集大但每次更新量小的场景。

选错输出模式是新手最容易踩的坑。比如在window聚合下你选了append，但没设置watermark，那么Spark会直接报错。因为它无法判断某个窗口什么时候算“完成”。搭配了watermark之后，append模式才能正常工作。

Sink方面，生产环境最常用的是Kafka和文件。

用Kafka sink时，你的输出DataFrame里必须有一个key字段或value字段，value字段必须是string类型。如果你聚合完之后列名不是value，需要用alias转一下。

val query = stats.writeStream .outputMode("update") .format("kafka") .option("kafka.bootstrap.servers", "kafka01:9092") .option("topic", "log-output") .option("checkpointLocation", "hdfs:///tmp/checkpoint/log-output") .start()

用文件sink时，适合落地成parquet或json，然后由下游数仓去读：

val query = stats.writeStream .outputMode("append") .format("parquet") .option("path", "hdfs:///user/dw/stream_log/") .option("checkpointLocation", "hdfs:///tmp/checkpoint/stream_log") .start()

注意一个细节：文件sink只支持append模式。因为update/complete模式下，已经写入的文件理论上不能被修改，Spark没有能力做upsert到文件系统上。

4.4 容错与checkpoint的细节

Structured Streaming的容错依赖checkpoint。checkpoint里存了当前流处理的元数据、offset进度、已提交的输出操作，以及聚合状态。重启作业时，只要指定同一个checkpointLocation，就能从上次的状态恢复。

这个机制看起来很好用，但有一个非常关键的约束：checkpointLocation不能换。一旦你换了checkpoint目录，作业会报错，因为Spark发现之前的元数据对不上。所以生产环境里，checkpoint目录的设计要一次定好，轻易别改。

如果你改了业务逻辑，比如改变了状态计算的字段、改变了输出Kafka的topic，这些变更也会导致checkpoint关联失败，因为checkpoint里记录了之前的操作计划。类似场景我遇到过几次，解决方式也很简单：换一个新的checkpointLocation，从最新的Kafka offset开始重放，或者从目标位置重置offset，然后重建状态。

5. 参数调优与生产经验

5.1 参数配置与内存调优

Spark Streaming作业的性能，很大程度上取决于并行度和内存配置。我整理了一份我常用的参数清单，大家可以当模板用。

如果你用的是Structured Streaming，那上面这些参数里，maxRatePerPartition和backpressure也是适用的，只不过这个参数要在Kafka数据源选项里设置。

背压机制我要多说一句。它的作用就是：当Spark发现处理不过来了，会自动降低拉取速率；当处理能力富余了，又会自动提升。这个功能对于稳定性非常重要。我有一次忘了开背压，Kafka里正好有积压的几百GB数据，结果Spark一启动就疯狂拉数据，直接把下游的HDFS写入压垮了。开启背压之后，系统会慢慢把积压的数据“消化”掉，不会一次冲垮下游。

5.2 状态后端与RocksDB

如果你的流任务有状态聚合（groupBy或者mapGroupsWithState），那状态存储是你必看的调优点。Spark 3.2之前，状态存储默认是内存中的哈希表，状态量大了以后很容易OOM。Spark 3.x引入了RocksDB状态后端，把状态持久化到磁盘，用内存做缓存，内存占用情况会好很多。

开启方法是在Spark配置里：

spark.sql.streaming.stateStore.providerClass=org.apache.spark.sql.execution.streaming.state.RocksDBStateStoreProvider

要注意的是，RocksDB本身就是一个小型嵌入式数据库，它需要一些磁盘IO。如果你的executor是部署在本地磁盘上，注意预留足够的空间。我见过一个集群，因为RocksDB的本地目录被打满，导致状态写入失败，任务不断重启。最后在Spark配置里给stateStore目录指定了一个容量更大的路径，才解决问题。

5.3 批次积压的排查思路

批次积压是最常见的生产事故。表现就是Spark UI里的Scheduling Delay越来越大，批处理时间逐步拉长，最后整个作业处于“半死不活”的状态。

我的排查步骤是：

1. 先看是不是输入速率激增。是Kafka消息量涨了，还是上游重试导致重复生产。前者需要加资源，后者要上下游协调。
2. 再看是不是某个stage执行特别慢。点开Spark UI，看每个stage的耗时分布。如果某个stage的task计算时间差异巨大，说明有数据倾斜，需要加salting或者重新分区。
3. 看GC情况。如果executor的GC时间占比很高，说明堆内内存紧张，考虑增加executor内存、降低每批数据量，或者开启offheap。
4. 看是否有背压生效。如果maxRatePerPartition没有正确生效，你会看到输入速率一直保持很高的值，这时候要检查是不是启动参数没有读到配置文件。

这个排查顺序基本能把80%的问题定位出来。剩下20%往往是下游依赖的问题，比如写Kafka时目标topic的partition数不够，或者写入HDFS时NameNode压力大。

6. 常见问题与排查技巧实录

6.1 几个典型的“翻车”现场

我把自己以前遇到过的、以及帮别人排查过的几类高频问题整理成了一张表：

6.2 一个经典的“非法状态”问题

Structured Streaming里你可能会遇到一个报错：

Illegal State: The boundary of state or the delta of state has been dropped by the state store

这个报错的意思是：状态存储里的数据因为TTL过期被清理掉了，但你的处理逻辑还期望它存在。这种问题通常出现在你设置了非常小的watermark，但是任务运行中途停了很久，然后恢复时Kafka里有大段积压数据，这些数据的事件时间和当前时间差得太大，导致窗口状态被清理了，而后续的补充数据还想更新这些窗口。

解决方式也很直接：调大watermark的宽容度，或者确保Kafka没有长时间积压，或者定期对consumer group做offset调整。这个报错在设计延迟补偿场景时特别容易踩到，建议提前评估。

6.3 监控与告警的落地

说实话，流作业不配监控，就相当于裸奔。我每上一个Spark Streaming任务，监控面板里至少要有这几个指标：

• 当前批次输入记录数和记录速率
• 每个批次的调度延迟、处理延迟
• 窗口聚合状态大小（RocksDB目录大小）
• Kafka消费lag（可以用Kafka自带的工具查）
• executor的GC耗时、内存使用率

如果你们的运维平台是自建的，可以考虑把Spark UI暴露的JMX指标接到监控看板；如果图省事，也可以写一个定期shell脚本，调用Spark的REST API获取application状态，把异常情况推到告警群里。

我自己习惯的做法是：在作业代码里用foreachBatch或者foreachRDD，把每个批次的统计信息（比如输入条数、处理耗时、结果行数）打出来，然后通过日志采集到统一的监控平台。这样业务层面的指标也齐全，定位问题时的上下文也够。

7. 经验总结与写在最后的建议

做流处理这几年，我用Spark Streaming处理过日志、埋点、订单、IoT传感器数据，踩过无数坑，也见过很多团队在DStream和Structured Streaming之间反复折腾。我的个人建议是：如果你是新项目，直接学Structured Streaming，别再花时间研究DStream了。它的API设计和Spark SQL无缝衔接，调优路径清晰，容错机制也更现代。如果你是接手老项目，那还是要踏踏实实把DStream的窗口、状态、checkpoint这套搞明白，因为那些老代码不会因为你不想用就消失。

最后再分享一个实操中容易被忽略的小技巧：任何流处理作业第一次上线时，都建议先用一个低的输入速率跑几天，让状态和checkpoint都稳定下来，再逐渐放开速率。别一上来就全量灌数据，否则你根本分不清是状态存储的问题、shuffle的问题，还是下游目标存储的问题。稳妥起步，比什么都强。

Spark Streaming不是银弹，但从“批”到“微批”再到“真正的流”的演进路线里，它是绕不开的一块基石。搞懂它、用好它，你在实时数据处理上的底盘就算稳了。