如果你做过实时链路一定被人问过一个问题要用 Spark Streaming 还是要上 Flink我先不直接给答案。先说个事实——很多团队在 Spark 3.x 上照样把实时看板、实时数仓做得稳稳的延迟做到几秒吞吐还不低。Spark Streaming 的核心价值从来不是单纯拼延迟而是让你用同一套 Spark 生态顺手把批处理和流处理一起解决掉。这篇指北是系列第四篇主要面向两类人一类是刚把 Spark 3.x 跑通、想往流式方向走的同学另一类是在 DStream 和 Structured Streaming 之间犹豫、不知道生产环境该押哪边的工程师。我不打算写一本手册而是把真正影响线上稳定性的东西——微批模型、Kafka 接入、offset 管理、容错语义、调优参数——按我自己的实践经验串起来。1. 流式处理的三个绕不开的问题与微批模型的答案1.1 延迟、吞吐、一致性先想清楚你要哪个流式引擎的设计里有一个三角约束延迟、吞吐、一致性。想要极低延迟通常要牺牲吞吐或一致性想要高吞吐往往得接受一定延迟。很多人一听到“流式”第一反应是“每条数据到了立刻处理”也就是真正的逐条事件驱动。像 Flink 这类引擎走的就是这条路延迟可以到毫秒级。而 Spark Streaming 从一开始选了另一条路微批Micro-Batch。微批的思路其实很朴素把持续不断流入的数据按固定时间间隔切成一批一批的小数据块每一批交给 Spark 的批处理引擎去执行。系统时间每过 1 秒就把这 1 秒内到达的数据封装成一个 RDD然后像跑普通 Spark 任务一样处理。所以从外部看Spark Streaming 是“准实时”不是真正的毫秒级实时。这个延迟的下限取决于你设置的批处理间隔一般生产环境常见的是 1 秒、2 秒、5 秒极限也能压到 500 毫秒左右但再往下就比较吃力了。这个取舍带来一个巨大的好处因为每一批本质上是 RDD所以 Spark 的容错、血缘、调度、资源管理这些批处理能力几乎原封不动地继承到了流式场景。你不需要维护两套引擎也不需要额外学一套复杂的分布式计算模型。对绝大多数业务来说比如实时大屏、数仓增量同步、异常监控1 到 5 秒的延迟完全够用。真正需要毫秒级的场景也有但比例比想象中低很多。先把这个三角约束想清楚后面很多选型就不会纠结。1.2 微批在 Spark 3.x 里到底改进了什么老版本里用 Spark Streaming很多人抱怨吞吐上不去一批数据稍微多点就调度不过来。这里有一部分原因是 Spark 2.x 时期的 shuffle 在流式场景下表现一般任务调度开销也比较大。到了 3.x情况改善了不少几个关键能力值得注意。第一个是统一 shuffle 服务Custom Shuffle Service的成熟。流式处理的每个批次都要做 shuffle批次之间不断重复如果 shuffle 服务不稳定整个流任务就会频繁失败。3.x 对这一块的稳定性和性能做了不少优化我自己维护的流任务在升到 3.x 之后shuffle 相关的失败明显变少。第二个是动态资源分配Dynamic Resource Allocation在流场景下的配合。流任务长期跑在集群上如果按峰值来申请资源低谷期就白白浪费。配合 YARN 的动态资源分配Executor 可以根据批次负载进行伸缩。不过要注意这个伸缩有延迟如果批处理间隔只有 1 秒资源频繁波动反而会导致不稳定所以生产上通常是批间隔较长时才开。第三个是预测执行Speculative Execution不再只是批处理的专利。某个 Executor 出现长尾拖慢了整个批次Spark 3.x 可以像批处理一样对卡住的任务进行推测执行。这一点在流式场景里很有用因为一个批次被拖住后续批次会积压严重时整个任务延迟报警。虽然默认配置比较保守但值得根据你的数据波动情况单独调一下。1.3 什么时候真的不适合微批我不建议为了凑数硬吹微批。如果你的业务是交易反欺诈、极高频行情推送、用户点击毫秒级响应这种场景用 Spark Streaming 是自找麻烦。微批模型天然存在“批次边界”的等待状态管理也不如专用流引擎精细做精确一次的状态更新会更绕。此外如果你不仅需要低延迟还需要复杂的事件时间乱序处理、动态水位线、多流 join 的精确语义Spark Streaming 也能做但写起来会更别扭不如直接用专门的流处理引擎。反过来讲如果你的场景是“数据从 Kafka 进来经过清洗聚合落到数仓或数据库延迟在秒级可接受”那 Spark Streaming 是非常务实的选择。它和 Spark SQL 共用了大量代码路径抛物线式的学习成本曲线比专用流引擎平滑得多。而且你如果团队里已经有一套 Spark 批处理体系流式任务和批处理任务可以共用同一套 UDF、同一套表结构定义不需要两拨人维护两套逻辑。这个好处经常被低估。2. Spark 3.x 选型DStream 与 Structured Streaming 的取舍2.1 DStream 的编程模型从来没有消失如果你去翻老项目还是会看到大量基于 DStream 的代码。DStreamDiscretized Stream就是一系列连续的 RDD每个 RDD 代表一个批次的数据。它提供的算子比如 map、flatMap、reduceByKey、window和 RDD 是同一套心智模型。val kafkaStream: InputDStream[(String, String)] KafkaUtils.createStream(...) kafkaStream .map(_._2) .flatMap(_.split( )) .map((_, 1L)) .reduceByKey(_ _) .foreachRDD { rdd // 在这里把结果写出去 }这段代码很好懂几乎就是把批处理里的 RDD 处理逻辑平移过来套上窗口操作就行。DStream 的优点是简单、直接底层是 RDD调试的时候可以看到每个批次的数据分布对 Open Source Spark 的老用户非常友好。但 DStream 的缺点也很明显你拿到的是 RDD而不是 DataFrame很多优化比如 Catalyst 的谓词下推、列式存储裁剪都用不上。窗口操作只能基于处理时间Processing Time做不了事件时间Event Time的乱序处理。如果你想做“事件发生在 5 分钟前、但数据迟到 2 分钟也能参与正确统计”这种需求DStream 会写到你怀疑人生。2.2 Structured Streaming 才是 Spark 3.x 的主航道如果你现在才开始一个新项目我的建议很简单直接用 Structured Streaming。Structured Streaming 的编程模型是把数据流看作一张无限增长的表格每一批新数据就是往这张表里追加行。你操作的是一张“流式 DataFrame”写起来跟 Spark SQL 几乎没有区别。val spark SparkSession.builder() .appName(streaming-orders) .master(yarn) .getOrCreate() val orders spark.readStream .format(kafka) .option(kafka.bootstrap.servers, host:9092) .option(subscribe, orders) .option(startingOffsets, latest) .load() .selectExpr(CAST(value AS STRING) as json) .selectExpr(get_json_object(json, $.userId) as userId, get_json_object(json, $.amount) as amount, get_json_object(json, $.ts) as ts)这段代码读完基本就掌握了 Structured Streaming 的输入方式。后续可以继续接 filter、groupBy、watermark、窗口聚合。输出端用writeStream可以指定 sink 为 Kafka、HDFS、JDBC 或控制台。Structured Streaming 真正强的地方在于它把“事件时间”和“水位线”做成了内建能力。比如你要统计每个用户每 5 分钟的下单金额即使数据乱序到达只要不超过水位线就能正确落入对应窗口。这背后是 Spark 会在状态存储里维护窗口的中间聚合结果超出 watermark 的窗口会被自动清理避免状态无限膨胀。输出模式有三种Append、Update、Complete。简单说Append 适合只追加结果的场景比如将聚合结果写入类似 Kafka 的消息队列Update 只输出发生变化的那部分结果适合实时更新数据库Complete 每次输出全量结果适合结果集比较小的场景。理解这三种模式比死记代码有用得多。2.3 如何选型一张表讲清楚对比维度DStreamStructured Streaming编程模型RDD 序列偏底层的算子操作DataFrame/DatasetSQL 友好事件时间处理不支持只能处理时间窗口原生支持 watermark 和事件时间窗口精确一次语义配合外部存储手动实现配合幂等输出或事务性输出更容易实现状态管理基于 updateStateByKeyAPI 较原始内置状态存储支持去重、聚合等与 Spark SQL 复用复用少UDF 需要单独处理高度复用可直接跑 SQL 逻辑生产案例占比3.x 新项目较少存量居多新项目主流我自己的原则是老代码没出大问题不折腾迁移但要控制新增逻辑不要在 DStream 上继续堆量新项目全部走 Structured Streaming。有一个例外情况是如果你的团队真的非常熟悉 RDD 编程且业务就是简单的 map/filter 后落库不需要窗口和状态那 DStream 的直观性也算一个优点。但这种情况太少大多数时候还是结构化流更值。3. Kafka 接入的完整实战offset、checkpoint 与链路稳定性3.1 从 Receiver 到 Direct为什么生产环境必须用 Direct 模式早些年 Spark Streaming 接 Kafka 用 Receiver 模式数据先被 Receiver 拉进 Executor 内存再写入 WAL。这个方案重复消费风险很高Executor 宕机时容易丢数据而且 Receiver 和 Spark 的并行度是脱节的扩容时坑很多。后来 Direct 模式成为标配Spark 直接和 Kafka 分区对应起来每个 Kafka 分区对应一个 Spark RDD 分区不再有 Receiver 中转。这意味着你的并行度天然对齐 Kafka 分区数不再被 Receiver 的线程池卡住。同时offset 的记录方式也变了Spark 会记录每个批次消费到哪个 offset为精确一次提供了基础。在 Spark 3.x 里Direct 模式就是默认且唯一的选择网上如果还有教你写 Receiver 的旧文档直接跳过。3.2 一个可复制的 Kafka 消费与窗口聚合 demo下面这段代码是我实际项目里常用结构删掉了业务字段保留骨架。注意几个关键点startingOffsets在无 checkpoint 时决定从哪里开始消费有 checkpoint 时它会直接忽略以 checkpoint 为准。failOnDataLoss默认是 true意思是如果 Spark 记录的 offset 已经不存在于 Kafka比如消息过期被清理任务会直接失败防止你悄悄丢数据。import org.apache.spark.sql.streaming.Trigger val query orders .withWatermark(ts, 2 minutes) .groupBy( window(col(ts), 5 minutes, 3 minutes), col(userId) ) .agg(sum(amount).as(total_amount)) .writeStream .outputMode(append) .trigger(Trigger.ProcessingTime(10 seconds)) .option(checkpointLocation, /spark/checkpoint/orders) .format(json) .option(path, /spark/result/orders) .start() query.awaitTermination()解释一下withWatermark它告诉 Spark“事件时间戳用 ts 字段我接受最大 2 分钟的迟到数据”。聚合窗口是 5 分钟滑动 3 分钟。每个批次来到时Spark 根据事件时间和当前 watermark 决定数据归属过期窗口不再更新。输出模式用 append 意味着窗口结束后才把最终结果写出去。如果你的结果需要实时更新用 update 模式配 JDBC sink 更合适。trigger这里的 10 秒指“处理触发间隔”也就是结构化流每个批次的调度间隔注意它不是数据处理延迟的绝对保证。如果前一批还没处理完下一批触发会被推后。所以触发间隔设多少取决于你的平均处理耗时和数据峰值不能拍脑袋。3.3 checkpoint 的两面性它能恢复任务也能让你踩大坑Streaming 任务能长期稳定运行靠的就是 checkpoint 机制。Structured Streaming 会把 offset 进度、状态数据、处理逻辑相关元数据写入 checkpoint 目录。任务挂掉之后重新拉起Spark 会从 checkpoint 恢复继续处理没做完的部分避免从头消费。但 checkpoint 有一个让我头疼过的问题它绑定的是代码逻辑。如果你的业务逻辑代码做了结构性修改比如算子顺序变了、状态操作的 key 结构变了、升级了 Spark 版本导致序列化结构不兼容从旧 checkpoint 恢复时大概率会抛异常。你可能会看到类似“unexpected checkpoint offset”或者序列化错误。网上很多人的第一反应是“重装一遍”但这里有个真相如果代码升级导致 checkpoint 不兼容你需要的不只是清理 checkpoint而是要明确清理 checkpoint 后 offset 从哪里恢复。如果你之前用的是 checkpoint 里默认记录的 offset清理后任务会从 startingOffsets 配置重新开始可能把历史数据重复消费一遍也可能跳过正在消费的数据。所以生产上我强烈建议不要完全依赖 checkpoint 来管理 offset。选一个外部存储比如关系型数据库或 Redis自己周期性地记录每个分区的当前 offset。任务恢复时先读取外部存储里的 offset再显式传递给 Kafka source这样代码怎么升级、checkpoint 怎么清理你都能控制消费起点。这段经验是我在线上吃过一次数据空窗的亏之后总结出来的。3.4 生产链路的稳定性自检清单无论你用 DStream 还是 Structured Streaming上线前按这个清单过一遍能减少大多数半夜被叫醒的概率数据源 Kafka 集群的 Topic 分区数是否固定如果 topic 分区数在中途调整offset 对应关系会变复杂下游写入目标是否具备幂等性比如数据库表有没有唯一键文件写入路径是否按批次分区有没有任务重启后的自动拉起机制是 YARN 的自动重试还是调度平台的 supervisor有没有监控批次处理耗时和丢弃记录数建议至少检查 processing time、scheduling delay 和 input rows per batch 这几个指标checkpoint 目录是否有可靠备份比如挂在分布式文件系统而非本地磁盘是否给流任务设置资源上限防止单个流任务把整个队列的 Executor 全占光。4. 把 exactly-once 落到实处容错模型与幂等输出4.1 容错的三根柱子checkpoint、WAL 与血缘很多刚接触流式计算的同学会把 exactly-once 当成一个“功能开关”其实它是一整套配合的结果。Spark Streaming 的容错基础有三层第一层是 checkpoint定期把状态和 offset 进度写入持久化存储第二层是 WALWrite Ahead Log用于 Receiver 时代的日志Direct 模式下更多是依赖 Kafka 本身的持久化第三层是 RDD 血缘一批数据如果处理到一半失败了Spark 可以通过血缘重新计算而不需要从 Kafka 重新消费。这三层结合的效果是只要 Kafka 里的数据还在你就能从失败中恢复重新计算被中断的批次。但请注意这里是“计算不丢”不等于“输出不重”。处理完一个批次并写入外部的过程中如果任务刚写完一部分就崩溃重启后会从这一个批次的起点重算之前已经成功写入的少量数据就会被再次写一遍。这就引出了流式系统里最难搞的重复问题。4.2 为什么 at-least-once 避不开理想情况下我们想要“每个数据恰好被处理一次恰好被输出一次”。但分布式系统里“处理结果已经写出去了”和“系统确认结果已经写出去了”之间有一个时间窗口。任务在写完数据库之前崩溃那重启后会重跑任务在写完数据库之后、还没来得及确认时崩溃重启后也会重跑。你无法从逻辑上区分这两种情况。所以纯靠引擎本身能保证的通常只有 at-least-once至少一次也就是不丢但可能重复。明白这一点你就不会去追求“依赖 Spark 自动实现精确一次”而是接受现实引擎保证你不丢重复由下游幂等来消化。这其实也是 Flink 等引擎在真正意义上实现 exactly-once 时要和 Kafka 的幂等 Producer、事务性输出配合的原因原理是相通的用事务让写入变成原子操作或利用唯一约束让重复写入无效。4.3 用幂等写出把重复变成无害幂等写出的核心是让“写两次”和“写一次”效果一模一样。以 JDBC 写数据库为例假设你按订单唯一键 userId 时间窗口写入统计结果建表时在聚合维度上建唯一索引写入用 UPDATE 而不是 INSERT。重复运行一遍数据不会翻倍只是覆盖成相同的值。CREATE TABLE order_stats ( user_id STRING, window_start TIMESTAMP, total_amount DECIMAL, PRIMARY KEY (user_id, window_start) );在 Structured Streaming 里你可以用 foreachBatch在每个微批里对 DataFrame 做“先删后插”或“按主键更新”。query.writeStream .foreachBatch { (batchDF: DataFrame, batchId: Long) batchDF.write .mode(overwrite) .jdbc(jdbcUrl, order_stats, props) } .option(checkpointLocation, /spark/checkpoint/orders) .start()这里有一个容易犯错的地方mode(overwrite)会覆盖整张表如果你同时跑多个任务或者周期重叠会互相覆盖。更稳妥的做法是在目标表里加一个批次标识或业务日期分区只覆盖当天分区不影响历史数据。另一种思路是把结果写回 Kafka利用 Kafka 的 key 做 upsert下游再根据 key 去重。无论你选哪种核心是一样的让重复执行变成无害执行。4.4 关于 exactly-once 的另一个常见误区有人会把“Kafka 里的 offset 提交”和“数据消费成功”混为一谈。Structured Streaming 中offset 提交到 checkpoint 是引擎自动做的但它不保证你的外部系统一定成功。比如你把数据写到了 HDFSHDFS 文件刚写了几个块Executor 挂了HDFS 上留下了不完整的临时文件。Spark 恢复后重算重新写文件如果输出路径是覆盖式最终文件是完整的但如果输出路径是 append 式就可能残留半截文件。所以生产上我更推荐把结果写到按批次号或时间戳命名的目录比如/result/dt2024-01-01/batch123/每个批次一个子目录。即使某个批次失败重跑也是重新写同一个子目录不会污染其他批次。等整个批次确认成功后再通过 HDFS 的原子目录切换或者命令把临时目录移动到最终位置。这套思路在各种实时数仓实践里非常常见本质就是利用文件系统的原子操作来弥补引擎层做不到的精确一次。5. 生产环境调优笔记并行度、背压与延迟排查5.1 并行度到底怎么设流式任务的并行度很多人喜欢照抄批处理经验上来就spark.sql.shuffle.partitions200也不管数据量。流式任务和批处理有个差别它的 input 是持续的每个批次的数据量由 Kafka 积压和消费速率决定。input 侧的并行度由 partition 数决定不要盲目改。假设你的 topic 有 6 个分区一个批次就 6 个任务去拉取这是合理的。如果你强行repartition(100)多出来的任务只是把数据重新洗了一遍除了增加 shuffle IO 没有任何收益。那什么情况才需要调一般出现在聚合场景。比如 groupBy 之后的聚合下游 shuffle 分区如果太少单任务处理的数据量过大批次处理时间会变长。这时去调spark.sql.shuffle.partitions才有意义。另外要注意 Executor 的 executor-cores要给每个 Executor 留至少 1 到 2 个核给后台任务和 GC别把所有核都占满否则批次调度和内存回收会互相打架。我习惯是单个 Executor 上 3 到 4 个 core配合 4 到 8 GB 内存再根据数据量调整并行度不要动不动就一个 Executor 挂十几个线程。5.2 背压和流量控制背压的目的是防止“上游数据暴涨下游处理不过来”导致一批挤着一批、延迟无限上涨。DStream 里的控制手段是spark.streaming.backpressure.enabledtrue配合spark.streaming.backpressure.rateEstimator来平滑消费速率。它通过限制每秒拉取的最大记录数间接控制每个批次的负载。最粗暴但也最有效的参数是spark.streaming.kafka.maxRatePerPartition即每个 Kafka 分区每秒钟最多消费多少条。Structured Streaming 里对应的思路是maxOffsetsPerTrigger这个参数很实用。它限制每次触发最多消费多少条数据避免刚重启时把积压的大批量数据全部拉进来导致雪崩。需要注意它不是“背压”的全部更像一个阀门。如果数据量持续超限你会看到处理速度稳定在阈值但延迟会慢慢增长。这其实不算坏事至少系统没有崩溃。要真正消化积压需要结合触发间隔、分区数、聚合复杂度一起调。参数DStreamStructured Streaming作用背压总开关spark.streaming.backpressure.enabled无直接对应靠触发机制控制是否根据处理速度动态调整消费速率每分区消费上限spark.streaming.kafka.maxRatePerPartitionmaxOffsetsPerTrigger限制单批摄入量触发间隔batchIntervaltrigger(ProcessingTime)多久触发一次微批分区对应Kafka 分区与 RDD 分区一一对应默认支持多分区并发读取控制并行度5.3 延迟上涨时第一件事其实很简单线上最常见的报警就是“任务处理延迟超过 XX”。很多人的第一反应是加资源但导致延迟上涨的原因通常是下面几种数据量没有变大但某个 Executor 上的任务出现数据倾斜个别任务耗时极长下游数据库连接数不够写操作排队批次被 Sink 阶段堵住同一批次里大量小文件写入 HDFSNameNode 的压力被顶上来频繁 GC日志里能看到 Full GC 的记录checkpoint 写入的存储集群抖动状态保存变慢。排查的顺序我建议这样先看每个批次的处理耗时Processing Time和调度延迟Scheduling Delay确认瓶颈是在计算还是调度再看单个任务的最大耗时和最小耗时差值判断有没有倾斜然后把 Sink 阶段单独测一下看看是不是下游扛不住。数据库连接池打满这个坑我印象特别深——曾经一个任务延迟从 5 秒涨到 20 分钟结果只是下游 MySQL 的最大连接数被另一个报表任务占满了跟 Spark 本身一点关系没有。所以排查时先看下游再看上游不要把锅全甩给引擎。调优这件事核心原则是先定位再动参数不要一上来就把并行度翻倍、把触发间隔调小。很多时候延迟上涨是因为某个环节打满盲目加资源只会让瓶颈换一个位置。最后说一点个人体会。我刚开始做流处理的时候特别执着于把延迟压到 1 秒以内结果反复调参、反复出问题。后来做稳定了才发现与其追求极致的低延迟不如先把“重启不丢数据、重复写入无害、积压不拖垮集群”这三件事做好。流任务跑几天几周不出事比那几百毫秒的提升重要太多。如果你也是刚上手 Spark Streaming我建议第一次放轻松一点用 5 秒的触发间隔跑起来把 offset、checkpoint、幂等输出这些链路打磨顺再慢慢往下压间隔。这座山没有想象中陡但前几步走得稳一点后面会省很多心。