外行聊流处理总喜欢把技术选型和架构吹得天花乱坠真回到代码层面很多人连 Structured Streaming 里一个窗口聚合的 watermark 都调不明白。我见过不少团队Kafka 数据都堆到磁盘了还在纠结 DStream 和 DataFrame 的区别。这篇文章就用实际跑过的例子把 Structured Streaming 从原理到生产环境里最容易被忽视的那些细节一次说清楚。你跟着把代码敲一遍再回头看这张标题应该就不会再说“不会”了。这篇内容适合三类人刚开始接触 Spark 流计算、想把手上的离线任务平滑迁到实时链路上的开发者已经在用 Structured Streaming但遇到数据延迟、丢状态、checkpoint 反复失效问题的运维和研发以及面试前想快速把流处理知识体系串一遍的人。我会直接从 API 演进讲起把事件时间、窗口、水印、sink 选择、参数调优和排查经验全部串起来。1. Structured Streaming 到底在解决什么问题1.1 从 RDD 思维到声明式思维早期用 Spark Streaming 写流处理核心对象是 DStream。DStream 本质上是一串不断产生的 RDD你在上面做 map、flatMap、reduceByKey写的时候感觉像是在操作一个无限序列的集合。这种模型的难点在于RDD 的转换逻辑和底层执行绑得太紧做窗口、做状态、做去重都需要自己维护外部存储代码量一大就失控。而且 DStream 的处理时间语义是“当前批次的系统时间”对事件时间的支持很弱业务想统计“用户真正下单那一刻”的数据几乎要自己绕很大一圈。Structured Streaming 最大的改变是把“流”重新抽象成“一张没有边界的表”。上面每来一条数据就是往表里插一行你写的 DataFrame 查询在这张表上每秒或每几百毫秒“重新执行”一次结果继续输出到下游。也就是说你不需要再学习另一套 API批处理里的 select、filter、groupBy、join在流式场景里完全复用。代码从一个自定义的 DStream 转换链变成了和离线 ETL 几乎一致的 DataFrame 操作这套统一 API 的收益远比想象中大。小提示如果你所在团队已经有成熟的批处理 Spark SQL 经验切换到 Structured Streaming 的学习成本很低。你真正需要重新理解的是时间语义和输出模式而不再是 DataSource API 本身。1.2 微批模型与连续处理之争Structured Streaming 默认的引擎模型是微批Micro-Batch模型。数据进来以后会等待一个触发间隔比如 500ms 或 1s攒成一小批交给 Spark 的调度引擎去跑。这么做的好处是天然继承了 Spark 的容错机制每批任务失败可以直接用 lineage 重算checkpoint 里存的是批次的偏移信息和状态快照恢复逻辑和离线任务一样可靠。代价是端到端延迟最低只能到百毫秒这个级别做不到毫秒级响应。Structured Streaming 后来也推出了 Continuous Processing 模式用连续算子替代小批次官方宣称可以达到毫秒级延迟但支持的算子受限状态操作和聚合场景不完整。在我自己接过的项目里连续处理模式基本只适合做简单的 filter 和 map 转换稍微复杂一点的需求还是得回到微批模型。所以我给团队选型时的判断标准通常是如果业务要求的是分钟级或秒级延迟Kafka 到消息产生再到最终结果可见微批模型完全够用如果业务真的需要毫秒级实时风控或交易监控那应该直接考虑 Flink 这类原生流引擎而不是在 Spark 上强行优化延迟。不要为了一个不明确的指标引入高复杂度方案。2. 核心 API 与五种 Sink 的选型2.1 readStream 数据源怎么写Structured Streaming 官方支持的数据源有四类Filejson、csv、parquet、orc、text、Kafka、Rate用于压测、Socket调试用。实际生产环境里Kafka 和文件源占了绝大多数场景。看一下典型的 Kafka 源写法Python 示例from pyspark.sql import SparkSession from pyspark.sql.functions import col, from_json, window from pyspark.sql.types import StructType, StructField, StringType, LongType spark SparkSession.builder \ .appName(structured_streaming_demo) \ .config(spark.sql.shuffle.partitions, 4) \ .getOrCreate() schema StructType([ StructField(user_id, StringType(), True), StructField(action, StringType(), True), StructField(ts, LongType(), True) ]) raw_df spark.readStream \ .format(kafka) \ .option(kafka.bootstrap.servers, localhost:9092) \ .option(subscribe, user_actions) \ .option(startingOffsets, latest) \ .option(maxOffsetsPerTrigger, 10000) \ .load() \ .select( from_json(col(value).cast(string), schema).alias(data) ) \ .select(data.*)这里面有两个点容易被忽略。第一value字段默认是二进制必须转成 string 再用from_json解析第二startingOffsets是“作业首次启动时的读取位置”如果 checkpoint 已经存在这个参数不会生效Kafka 会从 checkpoint 记录的偏移继续读。很多人改了startingOffsets发现不生效就是因为忽略了 checkpoint 的优先级。文件源则常用于增量读取某个目录下的新文件比如日志采集落地成 parquet 后持续消费file_df spark.readStream \ .format(parquet) \ .schema(schema) \ .load(/data/logs) \ .select(user_id, action, ts)文件源需要注意maxFilesPerTrigger和maxFileAge两个参数。前者控制每个批次最多读多少新文件主要是防止目录里堆积文件过多时第一批被撑爆后者控制文件多久之后算“过期”默认值较大如果目录里既有新文件又有历史残留文件建议按业务保留周期显式设置否则会把不相关的历史文件重新读进来。2.2 writeStream 的六条输出路径写完输入核心是输出。Structured Streaming 的 sink 有 console、memory、file、kafka、foreach、foreachBatch 六种。console 和 memory只在开发调试阶段用。console 直接把结果打印到终端memory 把结果存成临时表适合用 Spark SQL 手动查看最新结果。file按批次写 parquet、json 等文件适合输出到数据湖的增量目录。kafka把结果写回 Kafka 的指定 topic常用于多层级的实时管道。foreach逐行处理结果适合对接数据库、Redis 等第三方系统。foreachBatch每批结果作为一个 DataFrame 处理后统一输出是我在生产环境用得最多的一种。foreachBatch 为什么这么重要因为很多 sink 有“类型限制”。比如你要把实时聚合结果写到 MySQL如果用 foreach每个批次要逐条执行 insert连接管理、批量提交完全靠手写性能和可靠性都很纠结。foreachBatch 则让你拿到整个批次的 DataFrame可以做 repartition用df.write.jdbc批量写甚至在同一批里先写 MySQL 再写 Redis事务边界更可控def write_micro_batch(batch_df, batch_id): # 按业务键分区避免小文件过多 writer batch_df.repartition(4).write \ .mode(append) \ .format(jdbc) \ .option(url, jdbc:mysql://... ) \ .option(dbtable, user_action_summary) \ .option(user, xxx) \ .option(password, xxx) writer.save() stream result_df.writeStream \ .outputMode(append) \ .foreachBatch(write_micro_batch) \ .trigger(processingTime10 seconds) \ .option(checkpointLocation, /data/checkpoint/stream_demo) \ .start()这里要特别提醒foreachBatch 的batch_id是单调递增的批次编号你可以用它做很多事比如记录当前批次的执行时间、动态更新元数据但不要依赖它做跨批次的状态累计。另外 foreachBatch 里的batch_df是一次性的临时 DataFrame如果你想对它写入多个目标必须在同一个函数里顺序执行完不能在函数外缓存后延迟使用。2.3 输出模式 Append、Update、Complete 怎么选输出模式决定了“每个微批结束后结果集中哪些行要被发到 sink”。这个选择直接影响状态管理规模和下游语义是新手最容易出问题的地方。Append只把本次新产生的行发给 sink。适合无聚合的纯过滤、转换或者带 watermark 的聚合后新增窗口结果。注意没有 watermark 的聚合查询Append 模式是被引擎拒绝的因为引擎不知道什么时候“这一行才算结束”。Update只输出本次有更新的行旧值在 sink 里被覆盖。对聚合场景很友好可以配合 watermark 用也可以不用但状态会一直保留。Complete每个批次都把全量聚合结果输出一遍。适合小数据量的全局统计或者写回 Kafka/console 方便对比数据量大时输出成本极高。实际业务中我的选择建议是如果是“明细实时过滤转发”无脑用 Append如果是“按维度聚合后写消息队列或数据库”用 Update 或 foreachBatch如果是“读取全量累计快照给监控大屏”才考虑 Complete。Use Update 配 foreachBatch 可能是最常见的生产组合。3. 窗口聚合与 Watermark流处理的分水岭3.1 事件时间和处理时间必须分开Structured Streaming 最核心的时间概念有两个处理时间Processing Time和事件时间Event Time。处理时间指数据到达引擎时的系统时间事件时间指业务上真正发生的时刻比如用户在 App 里点下按钮的那一秒。大部分简单的实时统计如果直接用处理时间做窗口数据只要稍有一点延迟或乱序统计结果就会失真。比如用户 14:00:01 产生了一条日志日志系统经过 Kafka 缓冲到 14:00:05 才被 Spark 读取如果用处理时间做 5 分钟窗口这条数据会进 14:00-14:05 的窗口而不是 14:00 的窗口。对需要精确计费、监控大盘、活动数据的场景这种偏差不可接受。所以只要业务上能拿到事件时间就应该全部按事件时间计算。Structured Streaming 里的做法是读取数据后识别出代表事件时间的列用window做时间窗口再用withWatermark标记乱序容忍度。事件时间字段建议统一约定为 Long 类型的毫秒时间戳或标准时间字符串尽量避免跨字段转换。看一个带窗口和水印的完整示例from pyspark.sql.functions import col, from_unixtime, window event_df raw_df \ .withColumn(event_time, from_unixtime(col(ts) / 1000).cast(timestamp)) windowed_df event_df \ .withWatermark(event_time, 10 minutes) \ .groupBy( window(col(event_time), 5 minutes, 1 minute), col(user_id) ) \ .agg({action: count}) \ .select( col(window.start).alias(window_start), col(window.end).alias(window_end), col(user_id), col(count(action)).alias(action_count) )这里的window(col(event_time), 5 minutes, 1 minute)表示窗口长度是 5 分钟每 1 分钟滑动一次也就是每 1 分钟计算一次过去 5 分钟内的数据。窗口长度和滑动间隔决定了结果的细粒度。如果不需要滑动可以用window(col(event_time), 5 minutes)固定为 5 分钟的滚动窗口。窗口函数在引擎内部实现并不复杂它会把每条记录“投影”进它所属的所有窗口区间然后再按窗口做 groupBy。写代码的时候要特别注意窗口列是 struct 类型里面包含start和end很多人在 select 时忘记取出来输出就变成了一堆嵌套结构。3.2 Watermark 的真正含义Watermark 是 Structured Streaming 处理乱序和迟到的核心机制。withWatermark(event_time, 10 minutes)表达的意思是引擎“认为”一条事件时间为 T 的数据最多只可能迟到 10 分钟。比如当前已经读到的最大事件时间是 14:30:00那么 watermark 时刻就是 14:20:00所有事件时间小于 14:20:00 的数据都被判定为“太晚到达”不再参与窗口计算。这个“丢弃”的动作不是立即发生的。窗口的结束条件是“watermark 已经越过窗口的 end 时间”。比如一个窗口是 14:00-14:05watermark 要到 14:15 才会越过它届时这个窗口才算“关闭”才能输出最终结果并且释放窗口相关的状态。所以 watermark 的选择本质上是在“容忍迟到数据的能力”和“结果输出延迟”之间做平衡。设得太短迟到数据被过早丢弃统计偏低设得太长窗口状态需要保留很久内存占用变大而且结果输出时间会明显推后。对大多数业务我常用的经验值是最大允许延迟的 1.5 到 3 倍不要拍脑袋设一个 1 小时的 watermark 然后抱怨状态太大。另一个常见的误解是watermark 设置后迟到的数据依然可以更新已经关闭的窗口不行。一旦窗口关闭它的状态就从状态存储里清掉了后面再来的数据不管迟到多久都无法更新历史结果。如果业务真的需要“修正历史数据”频率很高那应该把它设计成“结果写 sink 后由下游用离线任务定期修复”而不是单纯拉长 watermark。3.3 状态清理与去重场景除了窗口聚合状态管理还涉及另外两类常见场景流式去重和流流 Join。dropDuplicates可以用来做基于事件时间的去重比如同一个 user_id 10 分钟内只记录一次点击写法如下deduped_df event_df \ .withWatermark(event_time, 10 minutes) \ .dropDuplicates([user_id, action, ts])这里 watermark 的目的是让引擎可以定期清理旧去重状态否则状态会无限增长。去重逻辑是在 watermark 允许的时间范围内只有重复的键才会被过滤掉一旦键对应的数据已经“老过”水印窗口新来的相同键数据会被当成新数据保留。所以如果你的业务要求“永远去重”单靠 watermark 做不了得用外部存储如 Redis 或 HBase 做持久化唯一键。状态大小是流式任务最容易被忽视的监控项。聚合维度太多、键基数太大都会导致 state store 文件急剧膨胀。生产环境要关注两个配置spark.sql.streaming.stateStore.maintenanceInterval用于控制状态清理任务的执行周期默认是 60 秒另外在 Spark 3.2 之后可以用spark.sql.streaming.aggregation.stateFormatVersion2启用新的状态管理格式一般建议保持这个默认开启。如果状态持续增长无法回落首先要检查是不是 watermark 没生效、窗口维度太高或者用了 Complete 模式导致每次全量输出。3.4 流流 Join 的限制Structured Streaming 支持流和流之间 join但限制非常多主要靠 watermark 约束状态范围。比如订单流和支付流要在事件时间上关联代码形如order_stream read_kafka(order_topic) pay_stream read_kafka(payment_topic) joined order_stream \ .withWatermark(order_time, 30 minutes) \ .join(pay_stream.withWatermark(pay_time, 30 minutes), \ expr(order_id pay_order_id AND pay_time order_time AND pay_time order_time interval 30 minutes))这里 join 条件里必须带上时间区间引擎才能确定左侧和右侧数据各自需要保留多久。缺少时间区间约束的等值 join引擎会直接报错。流流 Join 最典型的坑是右表的数据到达时间比左表晚但如果右表的 watermark 设置不对匹配结果可能永远出不来。调试时建议先用 10 分钟水印跑一把小数据确认 join 输出符合预期再逐步调整参数。4. 生产环境实战触发间隔、确认机制与排障清单4.1 Trigger 怎么配才合理trigger参数直接决定流作业的微批节奏。最常见的设置有两种# 每 10 秒触发一个微批 .writeStream.trigger(processingTime10 seconds) # 上一个批次结束后立即进入下一批也就是尽可能快地跑 .writeStream.trigger(processingTime0 seconds)processingTime0 seconds并不是真的“不等待”它表示每个批次结束之后马上重新评估如果新数据还没攒够可能产生很多空批次。这时候和maxOffsetsPerTrigger搭配就很重要它限制了每个批次最多读取多少 Kafka 偏移间接实现了背压控制。比如单个批次最多拉 50000 条Spark 就知道哪怕 Kafka 里积压了 100 万条也要分批消化避免把源端一次性拉垮。生产环境里另一个很容易踩的坑是spark.sql.streaming.minBatchesToRetain。它决定 checkpoint 里最多保留多少个历史批次的元数据默认值是 100。如果作业停了一个月再重启新的批次号会直接从旧批次往后延但中间的历史批次如果被清理掉了恢复时可能因为找不到某个偏移而失败。遇到这种情况不一定要调大这个参数更有效的方式是确认 checkpoint 存储空间足够且不要频繁手动删除 checkpoint 内容。4.2 Exactly-Once 与 Checkpoint 的双刃剑Structured Streaming 的端到端一致性依赖于 checkpoint 和 sink 两类机制。checkpoint 里记录消费偏移、状态信息、已提交的批次列表确保作业重启后能恢复到上次提交的状态。这要求 checkpoint 目录必须放在可靠的分布式存储上比如 HDFS 或 S3。不要放在本地磁盘本地的文件一旦丢失整个流任务的“记忆”就没了统计数据会从源头开始重算或者直接断掉。写入 Kafka sink 时引擎通过“预写事务日志”保证至少一次语义如果你希望端到端不重不丢还需要让下游的消费端能处理重复。很多团队把结果写回 Kafka 之后下游再用一个幂等键去重这是最通用的套路。写入文件 sink 时Structured Streaming 能保证每个批次输出的文件是幂等的不会出现同一个批次重复写两份前提是 checkpoint 稳定。这背后是提交机制在起作用每个批次先写出临时文件成功后再统一把临时文件原子地改名为正式文件。如果你手动中途 kill 任务或者 checkpoint 路径换掉了就可能留下临时文件残留资源被白占。所以清理临时目录时不要直接把 checkpoint 一键删除要确认任务已经完全 stop。foreach 写外部数据库时要做到真正不重复最稳妥的还是幂等写入。比如 MySQL 里用唯一键 insert ... on duplicate key update批次重放的时候重复数据只更新不新增达到效果。4.3 与 Kafka Source 配合时的易错点使用 Kafka 作为输入源有三个细节直接决定数据正确性。一是startingOffsets的生效语义。首次启动时有效恢复后以 checkpoint 为准。很多人改了这个参数想重新消费历史数据结果发现没变化就是因为 checkpoint 目录里存了旧的 offset。想重头消费正确做法是换一个新的 checkpoint 目录或者删除旧 checkpoint确认业务可接受。二是 Kafka 的group.id。Structured Streaming 内部并不像普通 Kafka Consumer 那样由客户端管理 offset它是自己把 offset 提交到 checkpoint 的所以不要试图通过设置 consumer group 来控制消费位点。你设置的kafka.group.id在这里更多是元数据标识最终以 checkpoint 为准。三是消息格式。Kafka 里的 value 可能是 Avro、JSON、Protobuf通常需要先用 schema 去解析。如果消息中有脏数据导致from_json返回 null整条记录会变成 null 行在过滤时很容易被忽略。调试阶段一定要打印几条原始数据确认 schema 字段是否匹配。4.4 高频问题速查表现象根本原因解决思路结果一直不输出处于 Append 模式且聚合没配 watermark聚合场景改用 update/complete 或配 watermarkwatermark 没生效事件时间列类型不是 timestamp用cast(timestamp)转换后再用结果重复sink 非幂等且任务曾重启换 foreachBatch 做幂等写入或加唯一键作业重启后全部重跑checkpoint 丢失/被清理检查 checkpoint 目录存储与配置单个批次处理很慢读取量过大shuffle 分区不足调小 maxOffsetsPerTrigger调大 shuffle 分区数窗口数据晚到但没更新watermark 已关闭了窗口按业务容忍度调大 watermark状态无限增长watermark 缺失或键基数过高检查聚合键设计必要时外部存储文件写出来有一堆小文件每批并行度太高foreachBatch 里 repartition 控制输出文件数Kafka 读取值全是 null忘记把二进制 value 转 string 解析value.cast(string)from_json本地调试每次都从头消费本地 checkpoint 未清理换新 checkpoint 目录或显式删除这张表是若干次线上事故后的复盘总结很多问题不是模型本身复杂而是 API 语义和配置没对齐。你可以在排查问题的时候反复对照。5. 结合实操的一些经验与心得最后聊几个我自己在项目里反复验证过的习惯不一定写在大文档里但能帮你少走弯路。一是调试阶段不要直接接 Kafka。先用 rate source 或 socket source 生成模拟数据把核心的窗口和水印逻辑跑通再换 Kafka 源。这种方式能快速隔离问题数据压力也小。二是本地跑通后第一次上生产建议用Trigger.Once。这个模式可以让流任务只执行一个批次然后退出非常适合用来做数据回填和全量重算。有人可能会觉得这不算“流式”但它真的是排查逻辑错误最有效的杀手锏你可以像跑批处理任务一样把历史数据全部灌进来确认输出结果和离线计算一致再开实时模式。三是每个批次跑多久、处理多少条可以通过查看lastProgress和status拿到监控项里优先看 processRowsPerSecond、inputRowsPerSecond 和 numInputRows。它们能直观反映任务有没有跟上数据速率。如果一个流任务长期输入速率大于处理速率Kafka 中积压会越来越大光调maxOffsetsPerTrigger只能缓解根本解法是找瓶颈通常是 shuffle 分区数、状态序列化开销或下游 sink 吞吐。四是不要在一个流的代码里同时维护多个输出模式比如既要 Append 又要 Complete会让状态和 checkpoint 变得混乱。建议一个流一个输出语义多个结果写到多个不同 topic 时用 foreachBatch 在 batchId 维度统一分发。Structured Streaming 学起来真不算难它的很多坑其实都集中在“时间语义、状态管理、输出模式”这三件事上。你把这三点想明白再结合一两次线上真实的告警基本就掌握八九成了。如果你现在正卡在某个问题上不妨先把 checkpoint 目录和 watermark 时间打出来看看。很多时候不是代码写错了而是你没有真正弄清楚引擎在那一批数据上到底做了什么。 
长河君 · 2026/10/14 7:03:51 · 阅读 1.2k Structured Streaming从原理到生产：窗口、Watermark与调优实践
外行聊流处理总喜欢把技术选型和架构吹得天花乱坠真回到代码层面很多人连 Structured Streaming 里一个窗口聚合的 watermark 都调不明白。我见过不少团队Kafka 数据都堆到磁盘了还在纠结 DStream 和 DataFrame 的区别。这篇文章就用实际跑过的例子把 Structured Streaming 从原理到生产环境里最容易被忽视的那些细节一次说清楚。你跟着把代码敲一遍再回头看这张标题应该就不会再说“不会”了。这篇内容适合三类人刚开始接触 Spark 流计算、想把手上的离线任务平滑迁到实时链路上的开发者已经在用 Structured Streaming但遇到数据延迟、丢状态、checkpoint 反复失效问题的运维和研发以及面试前想快速把流处理知识体系串一遍的人。我会直接从 API 演进讲起把事件时间、窗口、水印、sink 选择、参数调优和排查经验全部串起来。1. Structured Streaming 到底在解决什么问题1.1 从 RDD 思维到声明式思维早期用 Spark Streaming 写流处理核心对象是 DStream。DStream 本质上是一串不断产生的 RDD你在上面做 map、flatMap、reduceByKey写的时候感觉像是在操作一个无限序列的集合。这种模型的难点在于RDD 的转换逻辑和底层执行绑得太紧做窗口、做状态、做去重都需要自己维护外部存储代码量一大就失控。而且 DStream 的处理时间语义是“当前批次的系统时间”对事件时间的支持很弱业务想统计“用户真正下单那一刻”的数据几乎要自己绕很大一圈。Structured Streaming 最大的改变是把“流”重新抽象成“一张没有边界的表”。上面每来一条数据就是往表里插一行你写的 DataFrame 查询在这张表上每秒或每几百毫秒“重新执行”一次结果继续输出到下游。也就是说你不需要再学习另一套 API批处理里的 select、filter、groupBy、join在流式场景里完全复用。代码从一个自定义的 DStream 转换链变成了和离线 ETL 几乎一致的 DataFrame 操作这套统一 API 的收益远比想象中大。小提示如果你所在团队已经有成熟的批处理 Spark SQL 经验切换到 Structured Streaming 的学习成本很低。你真正需要重新理解的是时间语义和输出模式而不再是 DataSource API 本身。1.2 微批模型与连续处理之争Structured Streaming 默认的引擎模型是微批Micro-Batch模型。数据进来以后会等待一个触发间隔比如 500ms 或 1s攒成一小批交给 Spark 的调度引擎去跑。这么做的好处是天然继承了 Spark 的容错机制每批任务失败可以直接用 lineage 重算checkpoint 里存的是批次的偏移信息和状态快照恢复逻辑和离线任务一样可靠。代价是端到端延迟最低只能到百毫秒这个级别做不到毫秒级响应。Structured Streaming 后来也推出了 Continuous Processing 模式用连续算子替代小批次官方宣称可以达到毫秒级延迟但支持的算子受限状态操作和聚合场景不完整。在我自己接过的项目里连续处理模式基本只适合做简单的 filter 和 map 转换稍微复杂一点的需求还是得回到微批模型。所以我给团队选型时的判断标准通常是如果业务要求的是分钟级或秒级延迟Kafka 到消息产生再到最终结果可见微批模型完全够用如果业务真的需要毫秒级实时风控或交易监控那应该直接考虑 Flink 这类原生流引擎而不是在 Spark 上强行优化延迟。不要为了一个不明确的指标引入高复杂度方案。2. 核心 API 与五种 Sink 的选型2.1 readStream 数据源怎么写Structured Streaming 官方支持的数据源有四类Filejson、csv、parquet、orc、text、Kafka、Rate用于压测、Socket调试用。实际生产环境里Kafka 和文件源占了绝大多数场景。看一下典型的 Kafka 源写法Python 示例from pyspark.sql import SparkSession from pyspark.sql.functions import col, from_json, window from pyspark.sql.types import StructType, StructField, StringType, LongType spark SparkSession.builder \ .appName(structured_streaming_demo) \ .config(spark.sql.shuffle.partitions, 4) \ .getOrCreate() schema StructType([ StructField(user_id, StringType(), True), StructField(action, StringType(), True), StructField(ts, LongType(), True) ]) raw_df spark.readStream \ .format(kafka) \ .option(kafka.bootstrap.servers, localhost:9092) \ .option(subscribe, user_actions) \ .option(startingOffsets, latest) \ .option(maxOffsetsPerTrigger, 10000) \ .load() \ .select( from_json(col(value).cast(string), schema).alias(data) ) \ .select(data.*)这里面有两个点容易被忽略。第一value字段默认是二进制必须转成 string 再用from_json解析第二startingOffsets是“作业首次启动时的读取位置”如果 checkpoint 已经存在这个参数不会生效Kafka 会从 checkpoint 记录的偏移继续读。很多人改了startingOffsets发现不生效就是因为忽略了 checkpoint 的优先级。文件源则常用于增量读取某个目录下的新文件比如日志采集落地成 parquet 后持续消费file_df spark.readStream \ .format(parquet) \ .schema(schema) \ .load(/data/logs) \ .select(user_id, action, ts)文件源需要注意maxFilesPerTrigger和maxFileAge两个参数。前者控制每个批次最多读多少新文件主要是防止目录里堆积文件过多时第一批被撑爆后者控制文件多久之后算“过期”默认值较大如果目录里既有新文件又有历史残留文件建议按业务保留周期显式设置否则会把不相关的历史文件重新读进来。2.2 writeStream 的六条输出路径写完输入核心是输出。Structured Streaming 的 sink 有 console、memory、file、kafka、foreach、foreachBatch 六种。console 和 memory只在开发调试阶段用。console 直接把结果打印到终端memory 把结果存成临时表适合用 Spark SQL 手动查看最新结果。file按批次写 parquet、json 等文件适合输出到数据湖的增量目录。kafka把结果写回 Kafka 的指定 topic常用于多层级的实时管道。foreach逐行处理结果适合对接数据库、Redis 等第三方系统。foreachBatch每批结果作为一个 DataFrame 处理后统一输出是我在生产环境用得最多的一种。foreachBatch 为什么这么重要因为很多 sink 有“类型限制”。比如你要把实时聚合结果写到 MySQL如果用 foreach每个批次要逐条执行 insert连接管理、批量提交完全靠手写性能和可靠性都很纠结。foreachBatch 则让你拿到整个批次的 DataFrame可以做 repartition用df.write.jdbc批量写甚至在同一批里先写 MySQL 再写 Redis事务边界更可控def write_micro_batch(batch_df, batch_id): # 按业务键分区避免小文件过多 writer batch_df.repartition(4).write \ .mode(append) \ .format(jdbc) \ .option(url, jdbc:mysql://... ) \ .option(dbtable, user_action_summary) \ .option(user, xxx) \ .option(password, xxx) writer.save() stream result_df.writeStream \ .outputMode(append) \ .foreachBatch(write_micro_batch) \ .trigger(processingTime10 seconds) \ .option(checkpointLocation, /data/checkpoint/stream_demo) \ .start()这里要特别提醒foreachBatch 的batch_id是单调递增的批次编号你可以用它做很多事比如记录当前批次的执行时间、动态更新元数据但不要依赖它做跨批次的状态累计。另外 foreachBatch 里的batch_df是一次性的临时 DataFrame如果你想对它写入多个目标必须在同一个函数里顺序执行完不能在函数外缓存后延迟使用。2.3 输出模式 Append、Update、Complete 怎么选输出模式决定了“每个微批结束后结果集中哪些行要被发到 sink”。这个选择直接影响状态管理规模和下游语义是新手最容易出问题的地方。Append只把本次新产生的行发给 sink。适合无聚合的纯过滤、转换或者带 watermark 的聚合后新增窗口结果。注意没有 watermark 的聚合查询Append 模式是被引擎拒绝的因为引擎不知道什么时候“这一行才算结束”。Update只输出本次有更新的行旧值在 sink 里被覆盖。对聚合场景很友好可以配合 watermark 用也可以不用但状态会一直保留。Complete每个批次都把全量聚合结果输出一遍。适合小数据量的全局统计或者写回 Kafka/console 方便对比数据量大时输出成本极高。实际业务中我的选择建议是如果是“明细实时过滤转发”无脑用 Append如果是“按维度聚合后写消息队列或数据库”用 Update 或 foreachBatch如果是“读取全量累计快照给监控大屏”才考虑 Complete。Use Update 配 foreachBatch 可能是最常见的生产组合。3. 窗口聚合与 Watermark流处理的分水岭3.1 事件时间和处理时间必须分开Structured Streaming 最核心的时间概念有两个处理时间Processing Time和事件时间Event Time。处理时间指数据到达引擎时的系统时间事件时间指业务上真正发生的时刻比如用户在 App 里点下按钮的那一秒。大部分简单的实时统计如果直接用处理时间做窗口数据只要稍有一点延迟或乱序统计结果就会失真。比如用户 14:00:01 产生了一条日志日志系统经过 Kafka 缓冲到 14:00:05 才被 Spark 读取如果用处理时间做 5 分钟窗口这条数据会进 14:00-14:05 的窗口而不是 14:00 的窗口。对需要精确计费、监控大盘、活动数据的场景这种偏差不可接受。所以只要业务上能拿到事件时间就应该全部按事件时间计算。Structured Streaming 里的做法是读取数据后识别出代表事件时间的列用window做时间窗口再用withWatermark标记乱序容忍度。事件时间字段建议统一约定为 Long 类型的毫秒时间戳或标准时间字符串尽量避免跨字段转换。看一个带窗口和水印的完整示例from pyspark.sql.functions import col, from_unixtime, window event_df raw_df \ .withColumn(event_time, from_unixtime(col(ts) / 1000).cast(timestamp)) windowed_df event_df \ .withWatermark(event_time, 10 minutes) \ .groupBy( window(col(event_time), 5 minutes, 1 minute), col(user_id) ) \ .agg({action: count}) \ .select( col(window.start).alias(window_start), col(window.end).alias(window_end), col(user_id), col(count(action)).alias(action_count) )这里的window(col(event_time), 5 minutes, 1 minute)表示窗口长度是 5 分钟每 1 分钟滑动一次也就是每 1 分钟计算一次过去 5 分钟内的数据。窗口长度和滑动间隔决定了结果的细粒度。如果不需要滑动可以用window(col(event_time), 5 minutes)固定为 5 分钟的滚动窗口。窗口函数在引擎内部实现并不复杂它会把每条记录“投影”进它所属的所有窗口区间然后再按窗口做 groupBy。写代码的时候要特别注意窗口列是 struct 类型里面包含start和end很多人在 select 时忘记取出来输出就变成了一堆嵌套结构。3.2 Watermark 的真正含义Watermark 是 Structured Streaming 处理乱序和迟到的核心机制。withWatermark(event_time, 10 minutes)表达的意思是引擎“认为”一条事件时间为 T 的数据最多只可能迟到 10 分钟。比如当前已经读到的最大事件时间是 14:30:00那么 watermark 时刻就是 14:20:00所有事件时间小于 14:20:00 的数据都被判定为“太晚到达”不再参与窗口计算。这个“丢弃”的动作不是立即发生的。窗口的结束条件是“watermark 已经越过窗口的 end 时间”。比如一个窗口是 14:00-14:05watermark 要到 14:15 才会越过它届时这个窗口才算“关闭”才能输出最终结果并且释放窗口相关的状态。所以 watermark 的选择本质上是在“容忍迟到数据的能力”和“结果输出延迟”之间做平衡。设得太短迟到数据被过早丢弃统计偏低设得太长窗口状态需要保留很久内存占用变大而且结果输出时间会明显推后。对大多数业务我常用的经验值是最大允许延迟的 1.5 到 3 倍不要拍脑袋设一个 1 小时的 watermark 然后抱怨状态太大。另一个常见的误解是watermark 设置后迟到的数据依然可以更新已经关闭的窗口不行。一旦窗口关闭它的状态就从状态存储里清掉了后面再来的数据不管迟到多久都无法更新历史结果。如果业务真的需要“修正历史数据”频率很高那应该把它设计成“结果写 sink 后由下游用离线任务定期修复”而不是单纯拉长 watermark。3.3 状态清理与去重场景除了窗口聚合状态管理还涉及另外两类常见场景流式去重和流流 Join。dropDuplicates可以用来做基于事件时间的去重比如同一个 user_id 10 分钟内只记录一次点击写法如下deduped_df event_df \ .withWatermark(event_time, 10 minutes) \ .dropDuplicates([user_id, action, ts])这里 watermark 的目的是让引擎可以定期清理旧去重状态否则状态会无限增长。去重逻辑是在 watermark 允许的时间范围内只有重复的键才会被过滤掉一旦键对应的数据已经“老过”水印窗口新来的相同键数据会被当成新数据保留。所以如果你的业务要求“永远去重”单靠 watermark 做不了得用外部存储如 Redis 或 HBase 做持久化唯一键。状态大小是流式任务最容易被忽视的监控项。聚合维度太多、键基数太大都会导致 state store 文件急剧膨胀。生产环境要关注两个配置spark.sql.streaming.stateStore.maintenanceInterval用于控制状态清理任务的执行周期默认是 60 秒另外在 Spark 3.2 之后可以用spark.sql.streaming.aggregation.stateFormatVersion2启用新的状态管理格式一般建议保持这个默认开启。如果状态持续增长无法回落首先要检查是不是 watermark 没生效、窗口维度太高或者用了 Complete 模式导致每次全量输出。3.4 流流 Join 的限制Structured Streaming 支持流和流之间 join但限制非常多主要靠 watermark 约束状态范围。比如订单流和支付流要在事件时间上关联代码形如order_stream read_kafka(order_topic) pay_stream read_kafka(payment_topic) joined order_stream \ .withWatermark(order_time, 30 minutes) \ .join(pay_stream.withWatermark(pay_time, 30 minutes), \ expr(order_id pay_order_id AND pay_time order_time AND pay_time order_time interval 30 minutes))这里 join 条件里必须带上时间区间引擎才能确定左侧和右侧数据各自需要保留多久。缺少时间区间约束的等值 join引擎会直接报错。流流 Join 最典型的坑是右表的数据到达时间比左表晚但如果右表的 watermark 设置不对匹配结果可能永远出不来。调试时建议先用 10 分钟水印跑一把小数据确认 join 输出符合预期再逐步调整参数。4. 生产环境实战触发间隔、确认机制与排障清单4.1 Trigger 怎么配才合理trigger参数直接决定流作业的微批节奏。最常见的设置有两种# 每 10 秒触发一个微批 .writeStream.trigger(processingTime10 seconds) # 上一个批次结束后立即进入下一批也就是尽可能快地跑 .writeStream.trigger(processingTime0 seconds)processingTime0 seconds并不是真的“不等待”它表示每个批次结束之后马上重新评估如果新数据还没攒够可能产生很多空批次。这时候和maxOffsetsPerTrigger搭配就很重要它限制了每个批次最多读取多少 Kafka 偏移间接实现了背压控制。比如单个批次最多拉 50000 条Spark 就知道哪怕 Kafka 里积压了 100 万条也要分批消化避免把源端一次性拉垮。生产环境里另一个很容易踩的坑是spark.sql.streaming.minBatchesToRetain。它决定 checkpoint 里最多保留多少个历史批次的元数据默认值是 100。如果作业停了一个月再重启新的批次号会直接从旧批次往后延但中间的历史批次如果被清理掉了恢复时可能因为找不到某个偏移而失败。遇到这种情况不一定要调大这个参数更有效的方式是确认 checkpoint 存储空间足够且不要频繁手动删除 checkpoint 内容。4.2 Exactly-Once 与 Checkpoint 的双刃剑Structured Streaming 的端到端一致性依赖于 checkpoint 和 sink 两类机制。checkpoint 里记录消费偏移、状态信息、已提交的批次列表确保作业重启后能恢复到上次提交的状态。这要求 checkpoint 目录必须放在可靠的分布式存储上比如 HDFS 或 S3。不要放在本地磁盘本地的文件一旦丢失整个流任务的“记忆”就没了统计数据会从源头开始重算或者直接断掉。写入 Kafka sink 时引擎通过“预写事务日志”保证至少一次语义如果你希望端到端不重不丢还需要让下游的消费端能处理重复。很多团队把结果写回 Kafka 之后下游再用一个幂等键去重这是最通用的套路。写入文件 sink 时Structured Streaming 能保证每个批次输出的文件是幂等的不会出现同一个批次重复写两份前提是 checkpoint 稳定。这背后是提交机制在起作用每个批次先写出临时文件成功后再统一把临时文件原子地改名为正式文件。如果你手动中途 kill 任务或者 checkpoint 路径换掉了就可能留下临时文件残留资源被白占。所以清理临时目录时不要直接把 checkpoint 一键删除要确认任务已经完全 stop。foreach 写外部数据库时要做到真正不重复最稳妥的还是幂等写入。比如 MySQL 里用唯一键 insert ... on duplicate key update批次重放的时候重复数据只更新不新增达到效果。4.3 与 Kafka Source 配合时的易错点使用 Kafka 作为输入源有三个细节直接决定数据正确性。一是startingOffsets的生效语义。首次启动时有效恢复后以 checkpoint 为准。很多人改了这个参数想重新消费历史数据结果发现没变化就是因为 checkpoint 目录里存了旧的 offset。想重头消费正确做法是换一个新的 checkpoint 目录或者删除旧 checkpoint确认业务可接受。二是 Kafka 的group.id。Structured Streaming 内部并不像普通 Kafka Consumer 那样由客户端管理 offset它是自己把 offset 提交到 checkpoint 的所以不要试图通过设置 consumer group 来控制消费位点。你设置的kafka.group.id在这里更多是元数据标识最终以 checkpoint 为准。三是消息格式。Kafka 里的 value 可能是 Avro、JSON、Protobuf通常需要先用 schema 去解析。如果消息中有脏数据导致from_json返回 null整条记录会变成 null 行在过滤时很容易被忽略。调试阶段一定要打印几条原始数据确认 schema 字段是否匹配。4.4 高频问题速查表现象根本原因解决思路结果一直不输出处于 Append 模式且聚合没配 watermark聚合场景改用 update/complete 或配 watermarkwatermark 没生效事件时间列类型不是 timestamp用cast(timestamp)转换后再用结果重复sink 非幂等且任务曾重启换 foreachBatch 做幂等写入或加唯一键作业重启后全部重跑checkpoint 丢失/被清理检查 checkpoint 目录存储与配置单个批次处理很慢读取量过大shuffle 分区不足调小 maxOffsetsPerTrigger调大 shuffle 分区数窗口数据晚到但没更新watermark 已关闭了窗口按业务容忍度调大 watermark状态无限增长watermark 缺失或键基数过高检查聚合键设计必要时外部存储文件写出来有一堆小文件每批并行度太高foreachBatch 里 repartition 控制输出文件数Kafka 读取值全是 null忘记把二进制 value 转 string 解析value.cast(string)from_json本地调试每次都从头消费本地 checkpoint 未清理换新 checkpoint 目录或显式删除这张表是若干次线上事故后的复盘总结很多问题不是模型本身复杂而是 API 语义和配置没对齐。你可以在排查问题的时候反复对照。5. 结合实操的一些经验与心得最后聊几个我自己在项目里反复验证过的习惯不一定写在大文档里但能帮你少走弯路。一是调试阶段不要直接接 Kafka。先用 rate source 或 socket source 生成模拟数据把核心的窗口和水印逻辑跑通再换 Kafka 源。这种方式能快速隔离问题数据压力也小。二是本地跑通后第一次上生产建议用Trigger.Once。这个模式可以让流任务只执行一个批次然后退出非常适合用来做数据回填和全量重算。有人可能会觉得这不算“流式”但它真的是排查逻辑错误最有效的杀手锏你可以像跑批处理任务一样把历史数据全部灌进来确认输出结果和离线计算一致再开实时模式。三是每个批次跑多久、处理多少条可以通过查看lastProgress和status拿到监控项里优先看 processRowsPerSecond、inputRowsPerSecond 和 numInputRows。它们能直观反映任务有没有跟上数据速率。如果一个流任务长期输入速率大于处理速率Kafka 中积压会越来越大光调maxOffsetsPerTrigger只能缓解根本解法是找瓶颈通常是 shuffle 分区数、状态序列化开销或下游 sink 吞吐。四是不要在一个流的代码里同时维护多个输出模式比如既要 Append 又要 Complete会让状态和 checkpoint 变得混乱。建议一个流一个输出语义多个结果写到多个不同 topic 时用 foreachBatch 在 batchId 维度统一分发。Structured Streaming 学起来真不算难它的很多坑其实都集中在“时间语义、状态管理、输出模式”这三件事上。你把这三点想明白再结合一两次线上真实的告警基本就掌握八九成了。如果你现在正卡在某个问题上不妨先把 checkpoint 目录和 watermark 时间打出来看看。很多时候不是代码写错了而是你没有真正弄清楚引擎在那一批数据上到底做了什么。