说实话我第一次用Structured Streaming的时候也以为它只是把老版Spark Streaming改了个名字结果真上手之后才发现这套API背后的设计思路完全换了套玩法。做实时数据处理这几年从最早的DStream到现在的Structured Streaming踩过的坑够写满一个小本子。这篇博客我准备把平时用得最多的核心知识点、代码模板和排查经验一次性讲透看完你至少能直接拿它去写第一版作业。不管你是刚接触流处理还是在用Kafka接Spark做实时报表、实时特征计算这篇都适合你。我会尽量少讲教科书定义多讲实际代码和卡壳的地方。1. 先搞清楚Structured Streaming到底解决什么问题1.1 从批处理到流处理的思维转变很多人在接触Structured Streaming之前心里默认的流处理就是来一条数据处理一条处理完就扔掉。这个直觉其实和Structured Streaming的设计思想完全不同。它把流看成一张永远在增长的无界表每来一条数据就是往这张表里插入一行。你写的查询逻辑本质上和在批处理中对一张静态表做查询没什么区别唯一的变化是这张表会不断有新数据进来结果也是持续更新的。这种设计带来的最大好处是API统一。你不必为了流处理单独学一套DStream算子也不用纠结reduceByKey和updateStateByKey之间那些微妙的差别直接拿DataFrame那套逻辑来写流的处理逻辑就行。我在给团队做分享的时候常常说如果你已经会写Spark SQL批处理那么Structured Streaming的入门成本真的很低难的反而是后续的窗口、水位线、容错这些流处理特有的概念。import org.apache.spark.sql.types._ 嗯这个官方文档里的一句话让我印象很深Structured Streaming is not just streaming, its about treating streaming as continuous queries. 理解了这句话后面很多东西就容易串起来了。1.2 核心模型无界表Unbounded Table到底怎么理解无界表这个概念刚开始确实有点绕我换一个生活化的类比你就明白了。把流数据想象成一家奶茶店门口的排队小票每来一个顾客就打印一张小票丢进箱子里。这个箱子就是无界表——它一直在变大永远装不满。你的查询语句比如统计今天每种口味卖了多少杯相当于每隔一段时间就对这个箱子里的小票做一次汇总生成一张结果表。Structured Streaming的执行引擎会定期把新到达的数据new data以微批次的形式触发查询然后把结果追加或者更新到外部存储。它内部维护的偏移量offset就相当于记录箱子里的票看到哪一张了这样即使程序崩溃重启也能从上次的位置继续读不会丢数据。这里有个特别重要的实操点你定义查询的时候写的是一份逻辑计划而不是一份一次性执行的程序。引擎会把这份逻辑计划编译成增量的执行计划每次触发对增量数据进行计算最终把结果合并。所以在写代码的时候你不需要自己去考虑这次处理的是哪一段数据这个脏活累活引擎已经替你做了。2. 动手写第一个Structured Streaming程序2.1 环境准备与依赖配置先说环境。我默认你用的Spark版本是2.4以上甚至3.x因为Structured Streaming在2.2以后才算真正稳定2.4加入了持续处理模式虽然是实验特性3.0以后成为主力推荐。如果你的生产环境还在用Spark 2.1那我建议你申请升级否则有些API和内置函数用起来会处处别扭——我就曾经因为生产环境版本太低导致一个内置JSON解析函数没法用最后只能自己写UDF绕过去。Maven坐标就一个核心依赖别多加乱七八糟的东西dependency groupIdorg.apache.spark/groupId artifactIdspark-sql-kafka-0-10_2.12/artifactId version3.3.0/version /dependency注意artifactId里的_2.12是Scala版本号必须和你的spark-core版本保持一致否则运行时会抛出各种奇怪的类找不到异常。如果你用PySpark麻烦稍微少一点但也得确认pyspark版本和spark-sql-kafka的jar包匹配。在本地调试时我推荐直接用spark-shell --packages org.apache.spark:spark-sql-kafka-0-10_2.12:3.3.0启动一个带Kafka依赖的交互式环境边写边跑看结果。2.2 从Kafka读取数据的完整代码下面这段代码是我平时最常见的起步模板——从Kafka读JSON格式的日志数据解析之后做简单清洗和过滤。以Scala为例完整代码如下import org.apache.spark.sql.SparkSession import org.apache.spark.sql.types._ import org.apache.spark.sql.functions._ val spark SparkSession.builder() .appName(structured-streaming-demo) .master(local[2]) .config(spark.sql.shuffle.partitions, 2) .getOrCreate() spark.sparkContext.setLogLevel(WARN) // 1. 定义JSON的schema建议手动指定而不是用spark.read.json推断 val jsonSchema new StructType() .add(event_time, StringType) .add(user_id, StringType) .add(action, StringType) .add(amount, DoubleType) // 2. 从Kafka读取流数据 val inputDF spark.readStream .format(kafka) .option(kafka.bootstrap.servers, localhost:9092) .option(subscribe, user_actions) .option(startingOffsets, earliest) // 还有latest选项 .load() // 3. 把Kafka消息的value字段从字节数组转成JSON并解析 val parsedDF inputDF .selectExpr(CAST(key AS STRING), CAST(value AS STRING)) .select(from_json(col(value), jsonSchema).as(data)) .select(data.*) .filter(col(amount).isNotNull)这段代码看起来平平无奇但里面有几个细节值得单独拎出来讲。首先schema一定要手动指定不要省事靠推断。在流式场景下spark.read.json只在批处理里可用而from_json要求你提供一个明确的schema如果不想老老实实定义结构后面遇到字段类型对不上、解析失败的情况你就该哭了。其次startingOffsets选earliest还是latest要提前想清楚。调试阶段我一般用earliest因为Kafka里可能积压了不少测试数据能马上看到输出但生产环境新任务上线时往往选latest避免把历史几百万条数据全部消费一遍导致启动高峰。这个选择和你的业务语义强相关没有绝对的对错。再补充一个实战经验selectExpr(CAST(value AS STRING))这一步不能省。Kafka的value在DataFrame里是binary类型很多新手直接select(value)拿去做JSON解析结果反序列化出来一堆乱码或者报错。加上CAST是成本最低的解法。2.3 输出模式怎么选读数据只是第一步真正的难点在于把结果写出去。Structured Streaming提供了三种输出模式选错模式的结果往往是任务运行到一半直接报错或者输出结果和你预期的完全不一样。第一种是Append模式它只把新增的行写入结果表适用于过滤、投影这种老数据不会被更新的场景。注意如果你做了聚合操作默认是不能用Append模式的除非你配合了watermark——这个后面会讲到。第二种是Update模式它会把本次批次中新增和更新的行都输出。对于做计数、求和这类聚合计算这是最符合直觉的选择。比如统计每个用户的累计下单金额每次微批次跑完结果表中用户的金额被更新了Update模式就会把更新后的整行输出到sink。第三种是Complete模式每次触发都会把整个结果表全部输出适用于结果集比较小的场景比如全局计数。如果结果表有上千万行你还用Complete模式那每次输出产生的IO和下游压力足以拖垮整个链路。我在本地调试时最推荐的sink是console它能直接把结果打印在终端不需要搭任何外部组件val query parsedDF .writeStream .outputMode(append) .format(console) .option(truncate, false) .option(numRows, 20) .start() query.awaitTermination()提示truncate默认是true字段一多就会把结果截断显示调试时根本看不全数据建议直接关闭。这是我在前三十分钟踩得最冤枉的一个坑。3. 窗口操作与事件时间处理3.1 滚动窗口和滑动窗口的计算逻辑很多人写流处理作业绕不开的一个场景就是每5分钟统计一次用户的活跃数。这就涉及窗口操作。Structured Streaming里最常用的两个窗口类型是滚动窗口Tumbling Window和滑动窗口Sliding Window。滚动窗口的时间区间是固定的、不重叠的比如0~5分钟、5~10分钟、10~15分钟。计算起来很直观——每条数据只会落进一个窗口。对应的函数调用方式如下val windowedCounts parsedDF .withColumn(event_time, to_timestamp(col(event_time))) .withWatermark(event_time, 10 minutes) .groupBy( window(col(event_time), 5 minutes), col(user_id) ) .count()滑动窗口则多了第二个参数表示滑动步长。比如窗口长度10分钟步长5分钟那么窗口的起始时间每隔5分钟移动一次相邻窗口之间会有一部分重叠一条数据可能同时属于多个窗口。这里我要强调一个很多人容易搞反的地方window函数生成的是一个新的结构体列包含start和end两个字段它不是一个普通的字符串时间戳。你在groupBy里可以直接用它但如果要输出到下游通常需要取window.end字段作为业务统计时间。窗口计算背后其实是Spark SQL里的一个逻辑熟悉批处理的人会觉得一点都不陌生。真正让流处理变得难以驾驭的是下面的时间处理话题。3.2 watermark机制与迟到数据实时场景里数据乱序和迟到是常态。移动端网络抖动、服务端日志采集延迟、消息队列积压随便一个环节出问题数据到达Spark的时间就比事件实际发生的时间晚很多。如果没有机制兜底晚到的数据要么被丢弃要么导致之前算好的窗口结果反复变化。Structured Streaming引入的watermark水位线机制本质上是一个数据可晚到多久的约定。你声明withWatermark(event_time, 10 minutes)意思就是告诉引擎所有事件时间晚于当前最大事件时间减去10分钟的数据都会被归入对应的窗口参与计算而更早的、晚到得离谱的数据会被当做过期数据忽略掉对应的窗口也会被关闭并清出内存状态。这里有个细节值得仔细琢磨watermark是建立在事件时间上的不是处理时间。很多人以为10分钟watermark就是允许任务处理延迟10分钟其实不对。它的判断标准是当前已经处理过的数据里最大的event_time减去10分钟作为水位线。举个例子假设上次批次里你看到的最大事件时间是18:00那么水位线就是17:50。如果新来一条数据事件时间是17:45那它会因为早于17:50而被丢弃即使它在系统里只晚到了几秒钟。从代码上看watermark必须和聚合操作放在一起才能生效val result parsedDF .withColumn(event_time, to_timestamp(col(event_time))) .withWatermark(event_time, 10 minutes) .groupBy( window(col(event_time), 5 minutes), col(user_id) ) .count()注意watermark会延迟窗口的关闭时间。如果你想窗口一结束就立刻输出结果watermark设得越长结果输出越晚。比如5分钟窗口配10分钟watermark理论上最早也要到事件时间超过窗口结束10分钟之后那个窗口的结果才会最终确定。这在设计实时报表时非常关键——到底要多实时就需要在水位线的容忍度和输出延迟之间做权衡。3.3 窗口聚合中的输出模式与条件限制窗口聚合搭配Append模式会让很多新手困惑。官方文档的解释是在加了watermark之后Append模式可用于聚合查询因为引擎能够保证一旦一个窗口的结果确定后续不会再更新。所以Append模式输入的是那些已经确定不再变化的窗口结果。Update模式就没有这个限制它可以在每次触发时输出新增或者更新的窗口结果。因此当你的下游系统能够处理更新语义时比如写到可更新的MySQL表用Update模式往往比Append模式更直观。我个人的实际选型经验是如果下游只允许追加写入比如HDFS目录、Kafka优先Append模式搭配足够长的watermark如果下游是支持更新的OLTP数据库或者能覆盖写的数据湖Update模式更舒服调试也方便。4. 状态管理与容错机制4.1 状态存储与状态超时清理有状态的计算比如累计活跃用户数每个用户的会话时长需要引擎在不同批次之间保存中间状态。Structured Streaming内部用状态存储来保存这些中间结果默认情况下状态存储在executor的内存和本地磁盘中由状态管理器定期checkpoint。之前我提到watermark会关闭窗口并清除状态这是状态清理的一种方式。但如果你做的不是窗口聚合而是普通的mapGroupsWithState或者flatMapGroupsWithState自定义状态处理那情况就复杂一些了。以mapGroupsWithState为例你需要自己定义状态的更新逻辑还要定义状态超时时间。官方提供了两种超时策略ProcessingTimeTimeout和EventTimeTimeout。前者以系统当前时间作为超时判断依据适合会话超时这类场景后者以事件时间作为依据和时间窗口语义保持一致。import org.apache.spark.sql.streaming._ val query parsedDF .groupByKey(row row.getAs[String](user_id)) .mapGroupsWithState(GroupStateTimeout.EventTimeTimeout)(updateState) def updateState( userId: String, inputs: Iterator[Row], state: GroupState[UserState] ): UserState { // 自定义状态逻辑 }做自定义状态时我强烈建议你在代码里显式处理state.hasTimedOut的情况否则超时之后旧状态一直滞留内存占用会肉眼可见地往上涨最后任务OOM。这个坑我在做用户行为分析时踩过一次排查了一整天才发现是超时分支没处理。4.2 checkpoint与端到端一致性Structured Streaming最核心的容错机制是checkpoint。你需要给每个query指定一个checkpoint位置通常是HDFS路径或者S3路径。引擎会把读进度offset、状态数据、已提交的输出批次等元数据都写到这个目录。val query parsedDF .writeStream .format(parquet) .outputMode(append) .option(checkpointLocation, /tmp/ss-checkpoint/user_actions) .option(path, /tmp/ss-output/user_actions) .start()很多人会问checkpoint到底保证什么样的语义答案是如果sink具有幂等写入能力Structured Streaming能实现端到端的exactly-once语义。具体来说引擎通过WAL预写日志记录每个批次要处理的offset范围在重启时从WAL恢复确保同一批数据不会因为崩溃而重复读取输出阶段它跟踪已经提交的批次ID并把批次ID传给sink方便下游做幂等去重。这里我想分享一个实际运维中特别容易忽视的点checkpoint路径不要随便换也不要多个query共用同一个checkpointLocation。一旦改动checkpoint路径引擎会认为这是一个全新的查询从头启动之前的offset记录全部失效结果可能从latest开始消费导致数据断层。我有一次因为部署脚本里的路径变量写错导致整个流任务重置白白丢了一个小时的增量数据被下游业务方追着问了半天。4.3 从故障中恢复的正确姿势要是程序真的崩了重启的流程也有讲究。正确的流程是尽可能保持代码逻辑不变只修复你要修复的部分仍然使用同一套checkpoint路径启动。引擎会自动从上次提交的位置继续处理。但如果你改动了查询的输出模式、source配置、聚合逻辑等影响执行计划的配置Spark很可能会抛出一个query plan not compatible之类的异常拒绝启动。这是很多人在开发环境改了代码之后突然发现任务起不来的原因。遇到这种情况成熟的处理方式有两种一是接受数据语义可能变化干脆换一个新的checkpoint路径从头消费二是走真正的升级流程确保新的查询计划能在旧的状态基础上继续不过这一般只支持很小的改动。5. 常见问题排查与性能调优5.1 背压机制与批次间隔怎么设置在Structured Streaming之前的Spark Streaming时代背压需要手动开启之后会自动根据处理耗时调节消费速率。而Structured Streaming相对简单它的触发间隔由trigger控制没有复杂的背压选项。但恰恰是这个简单让很多人忽略了它的存在。默认情况下Structured Streaming会尽可能快地处理也就是前一批次处理完就立刻开始下一批。这样能获得最低的延迟但如果数据源写入速率波动很大会造成处理压力忽高忽低。如果想要固定节奏可以定义ProcessingTime触发器val query parsedDF .writeStream .trigger(Trigger.ProcessingTime(10 seconds)) .format(console) .start()设置一个10秒的processing time trigger之后引擎会每10秒钟检查一次是否有新数据到达并且发起一个新的微批次。这相当于给了下游一个相对稳定的写入节奏。如果你的下游是分析型数据库或者某个不支持高并发写入的存储系统我建议务必设置一个trigger而不是让任务全功率运行否则下游被打爆是早晚的事。5.2 数据倾斜的排查思路Structured Streaming作业最常见的性能杀手不是别的而是数据倾斜尤其在做分组聚合的时候。假设你按user_id分组统计某个头部用户的日志量占全部日志的80%那么处理这个分区的executor会成为瓶颈其他executor空闲等待整个微批次处理时间被拖得很长。排查手段其实和批处理类似。第一看Spark UI中每个task处理的时间分布如果少数task耗时是其他task的几倍甚至几十倍基本可以断定发生了倾斜。第二看executor的内存和GC情况倾斜分区的executor往往内存压力大GC频繁。解决办法也直接。如果是按key分组的聚合可以先加一层随机前缀做二次聚合然后把结果合并。这个方案在批处理里是常规操作流处理里同样适用只不过需要注意加了前缀之后状态管理的key数量会变多状态存储的压力会相应增加。你得在真正动手之前评估一下这个trade-off权衡。对于写sink的时候发生的倾斜比如目标Kafka分区数少于Spark分区数可以开spark.sql.streaming.schemaInference相关的调优参数或者用repartition重新分区保持写出的分区数和下游分区数匹配。5.3 我踩过的几个典型坑第一个坑是Kafka的value值为null。不知道你们有没有遇见过Kafka里的墓碑消息tombstone recordvalue是nullJSON解析还好但一旦你把它当成字符串去split或格式化整个executor就会报错。解决方式是读取之后加一个过滤filter(col(value).isNotNull)。第二个坑是小文件问题。用Structured Streaming写Parquet到HDFS时如果微批次间隔很短、每个批次的数据量又很小生成的Parquet小文件会越积越多之后跑批任务扫描这些文件时目录列表和元数据开销会非常吓人。实践中应对方法有三个层次一是适当增大trigger间隔二是在写之前用coalesce控制输出分区数三是定期跑小文件合并任务。第三个坑是时区问题。事件时间从字符串转Timestamp时如果不显式指定时区Spark默认按照系统时区转化而系统时区通常是UTC。如果你的业务数据事件时间是东八区不处理的话窗口边界会整体偏移8小时统计出来的数据全都不在预期窗口里。排查这类问题时发现数据偏了八个小时十有八九就是时区问题优先检查spark.sql.session.timeZone配置。5.4 一个完整的流任务调试流程建议实操记录调试流任务跟调试批处理任务不太一样因为它是一个持续运行的程序不好直接打断点。我自己的做法是先准备一份历史数据的批量样本也就是把Kafka里最近的几万条数据落成一个临时文件然后用批处理的DataFrame API去模拟流式的处理逻辑。这样做的好处是如果逻辑有问题你在批处理里就能快速发现完全不用等流任务跑起来之后再去翻日志。等批处理逻辑验证通过再用readStreamwriteStream实际跑起来。第一次跑的时候强烈建议直接用console输出先跑个10分钟确认数据格式和结果符合预期然后再把输出源切换到Parquet、Kafka或者下游数据库。很多新人上来就直接往生产sink写结果数据格式不对下游表直接写入失败然后排查了半天才发现是源头数据里多了个隐藏字符。调试时的另一个好习惯是加spark-shell一行一行地执行。比如先读一个批处理DataFrame看看schema再试着调用from_json解析每一行都确认结果最后再把所有逻辑组合成一个copy-paste的脚本。这比一上来就写一整个.scala文件效率高得多因为编译期和运行期的报错信息在shell里更加直观。6. 进阶技巧与学习路线建议6.1 持续处理模式Continuous Processing到底能不能用Structured Streaming在2.3版本中引入了持续处理模式号称能做到毫秒级延迟不再是以微批次为单位而是以连续的方式处理。当时看到这个特性的时候我挺兴奋但真到了生产环境我依然选择微批次模式。原因很简单持续处理模式至今在很多场景中仍是实验性的支持的算子有限、端到端的exactly-once保证不如微批次成熟、状态存储的配合也有不少限制。如果你不需要超低延迟比如业务能接受1秒以上的延迟我建议继续用默认的微批次模式靠谱是第一位的。如果业务确实需要毫秒级延迟那你也该考虑是不是该用更专业的流处理引擎了而不是在Spark生态里硬扛。6.2 和Kafka、数据湖结合的完整架构参考最后说一个我在实际项目中比较常用的技术栈组合。数据源是业务日志通过Kafka接入Structured Streaming负责实时ETL和聚合结果写入Kafka供下游实时服务使用同时备份一份到数据湖用于离线分析。这套架构的好处是实时链路和离线链路共用同一套逻辑只需要把readStream换成read批处理版本几乎不需要改动。这种同一套代码批流一体的能力是我觉得Structured Streaming最有价值的地方。你不用维护两套逻辑也不用手动对账实时数据和离线数据因为它们的计算逻辑本来就是同一份。对于数据团队来说少一个对不齐的口径就少一个扯皮的源头。我也一直在关注社区的最新进展比如Spark 3.x对流处理的持续优化。但不管版本怎么更新核心的模型和思路是稳定的。你花时间吃透这套无界表模型、watermark机制、状态管理和检查点恢复后续版本升级充其量只是让你用上更顺手的工具。最后的最后如果让我给从事流处理开发的同行一个建议那就是别停留在会调API的层面遇到问题的时候多去读一读Spark执行的日志和UI多想想数据从源头到sink的整个生命周期很多所谓的神秘问题其实只是某个环节没有把控好而已。 
长河君 · 2026/10/14 7:03:51 · 阅读 1.2k Structured Streaming实战：从Kafka实时数据处理到底层机制
说实话我第一次用Structured Streaming的时候也以为它只是把老版Spark Streaming改了个名字结果真上手之后才发现这套API背后的设计思路完全换了套玩法。做实时数据处理这几年从最早的DStream到现在的Structured Streaming踩过的坑够写满一个小本子。这篇博客我准备把平时用得最多的核心知识点、代码模板和排查经验一次性讲透看完你至少能直接拿它去写第一版作业。不管你是刚接触流处理还是在用Kafka接Spark做实时报表、实时特征计算这篇都适合你。我会尽量少讲教科书定义多讲实际代码和卡壳的地方。1. 先搞清楚Structured Streaming到底解决什么问题1.1 从批处理到流处理的思维转变很多人在接触Structured Streaming之前心里默认的流处理就是来一条数据处理一条处理完就扔掉。这个直觉其实和Structured Streaming的设计思想完全不同。它把流看成一张永远在增长的无界表每来一条数据就是往这张表里插入一行。你写的查询逻辑本质上和在批处理中对一张静态表做查询没什么区别唯一的变化是这张表会不断有新数据进来结果也是持续更新的。这种设计带来的最大好处是API统一。你不必为了流处理单独学一套DStream算子也不用纠结reduceByKey和updateStateByKey之间那些微妙的差别直接拿DataFrame那套逻辑来写流的处理逻辑就行。我在给团队做分享的时候常常说如果你已经会写Spark SQL批处理那么Structured Streaming的入门成本真的很低难的反而是后续的窗口、水位线、容错这些流处理特有的概念。import org.apache.spark.sql.types._ 嗯这个官方文档里的一句话让我印象很深Structured Streaming is not just streaming, its about treating streaming as continuous queries. 理解了这句话后面很多东西就容易串起来了。1.2 核心模型无界表Unbounded Table到底怎么理解无界表这个概念刚开始确实有点绕我换一个生活化的类比你就明白了。把流数据想象成一家奶茶店门口的排队小票每来一个顾客就打印一张小票丢进箱子里。这个箱子就是无界表——它一直在变大永远装不满。你的查询语句比如统计今天每种口味卖了多少杯相当于每隔一段时间就对这个箱子里的小票做一次汇总生成一张结果表。Structured Streaming的执行引擎会定期把新到达的数据new data以微批次的形式触发查询然后把结果追加或者更新到外部存储。它内部维护的偏移量offset就相当于记录箱子里的票看到哪一张了这样即使程序崩溃重启也能从上次的位置继续读不会丢数据。这里有个特别重要的实操点你定义查询的时候写的是一份逻辑计划而不是一份一次性执行的程序。引擎会把这份逻辑计划编译成增量的执行计划每次触发对增量数据进行计算最终把结果合并。所以在写代码的时候你不需要自己去考虑这次处理的是哪一段数据这个脏活累活引擎已经替你做了。2. 动手写第一个Structured Streaming程序2.1 环境准备与依赖配置先说环境。我默认你用的Spark版本是2.4以上甚至3.x因为Structured Streaming在2.2以后才算真正稳定2.4加入了持续处理模式虽然是实验特性3.0以后成为主力推荐。如果你的生产环境还在用Spark 2.1那我建议你申请升级否则有些API和内置函数用起来会处处别扭——我就曾经因为生产环境版本太低导致一个内置JSON解析函数没法用最后只能自己写UDF绕过去。Maven坐标就一个核心依赖别多加乱七八糟的东西dependency groupIdorg.apache.spark/groupId artifactIdspark-sql-kafka-0-10_2.12/artifactId version3.3.0/version /dependency注意artifactId里的_2.12是Scala版本号必须和你的spark-core版本保持一致否则运行时会抛出各种奇怪的类找不到异常。如果你用PySpark麻烦稍微少一点但也得确认pyspark版本和spark-sql-kafka的jar包匹配。在本地调试时我推荐直接用spark-shell --packages org.apache.spark:spark-sql-kafka-0-10_2.12:3.3.0启动一个带Kafka依赖的交互式环境边写边跑看结果。2.2 从Kafka读取数据的完整代码下面这段代码是我平时最常见的起步模板——从Kafka读JSON格式的日志数据解析之后做简单清洗和过滤。以Scala为例完整代码如下import org.apache.spark.sql.SparkSession import org.apache.spark.sql.types._ import org.apache.spark.sql.functions._ val spark SparkSession.builder() .appName(structured-streaming-demo) .master(local[2]) .config(spark.sql.shuffle.partitions, 2) .getOrCreate() spark.sparkContext.setLogLevel(WARN) // 1. 定义JSON的schema建议手动指定而不是用spark.read.json推断 val jsonSchema new StructType() .add(event_time, StringType) .add(user_id, StringType) .add(action, StringType) .add(amount, DoubleType) // 2. 从Kafka读取流数据 val inputDF spark.readStream .format(kafka) .option(kafka.bootstrap.servers, localhost:9092) .option(subscribe, user_actions) .option(startingOffsets, earliest) // 还有latest选项 .load() // 3. 把Kafka消息的value字段从字节数组转成JSON并解析 val parsedDF inputDF .selectExpr(CAST(key AS STRING), CAST(value AS STRING)) .select(from_json(col(value), jsonSchema).as(data)) .select(data.*) .filter(col(amount).isNotNull)这段代码看起来平平无奇但里面有几个细节值得单独拎出来讲。首先schema一定要手动指定不要省事靠推断。在流式场景下spark.read.json只在批处理里可用而from_json要求你提供一个明确的schema如果不想老老实实定义结构后面遇到字段类型对不上、解析失败的情况你就该哭了。其次startingOffsets选earliest还是latest要提前想清楚。调试阶段我一般用earliest因为Kafka里可能积压了不少测试数据能马上看到输出但生产环境新任务上线时往往选latest避免把历史几百万条数据全部消费一遍导致启动高峰。这个选择和你的业务语义强相关没有绝对的对错。再补充一个实战经验selectExpr(CAST(value AS STRING))这一步不能省。Kafka的value在DataFrame里是binary类型很多新手直接select(value)拿去做JSON解析结果反序列化出来一堆乱码或者报错。加上CAST是成本最低的解法。2.3 输出模式怎么选读数据只是第一步真正的难点在于把结果写出去。Structured Streaming提供了三种输出模式选错模式的结果往往是任务运行到一半直接报错或者输出结果和你预期的完全不一样。第一种是Append模式它只把新增的行写入结果表适用于过滤、投影这种老数据不会被更新的场景。注意如果你做了聚合操作默认是不能用Append模式的除非你配合了watermark——这个后面会讲到。第二种是Update模式它会把本次批次中新增和更新的行都输出。对于做计数、求和这类聚合计算这是最符合直觉的选择。比如统计每个用户的累计下单金额每次微批次跑完结果表中用户的金额被更新了Update模式就会把更新后的整行输出到sink。第三种是Complete模式每次触发都会把整个结果表全部输出适用于结果集比较小的场景比如全局计数。如果结果表有上千万行你还用Complete模式那每次输出产生的IO和下游压力足以拖垮整个链路。我在本地调试时最推荐的sink是console它能直接把结果打印在终端不需要搭任何外部组件val query parsedDF .writeStream .outputMode(append) .format(console) .option(truncate, false) .option(numRows, 20) .start() query.awaitTermination()提示truncate默认是true字段一多就会把结果截断显示调试时根本看不全数据建议直接关闭。这是我在前三十分钟踩得最冤枉的一个坑。3. 窗口操作与事件时间处理3.1 滚动窗口和滑动窗口的计算逻辑很多人写流处理作业绕不开的一个场景就是每5分钟统计一次用户的活跃数。这就涉及窗口操作。Structured Streaming里最常用的两个窗口类型是滚动窗口Tumbling Window和滑动窗口Sliding Window。滚动窗口的时间区间是固定的、不重叠的比如0~5分钟、5~10分钟、10~15分钟。计算起来很直观——每条数据只会落进一个窗口。对应的函数调用方式如下val windowedCounts parsedDF .withColumn(event_time, to_timestamp(col(event_time))) .withWatermark(event_time, 10 minutes) .groupBy( window(col(event_time), 5 minutes), col(user_id) ) .count()滑动窗口则多了第二个参数表示滑动步长。比如窗口长度10分钟步长5分钟那么窗口的起始时间每隔5分钟移动一次相邻窗口之间会有一部分重叠一条数据可能同时属于多个窗口。这里我要强调一个很多人容易搞反的地方window函数生成的是一个新的结构体列包含start和end两个字段它不是一个普通的字符串时间戳。你在groupBy里可以直接用它但如果要输出到下游通常需要取window.end字段作为业务统计时间。窗口计算背后其实是Spark SQL里的一个逻辑熟悉批处理的人会觉得一点都不陌生。真正让流处理变得难以驾驭的是下面的时间处理话题。3.2 watermark机制与迟到数据实时场景里数据乱序和迟到是常态。移动端网络抖动、服务端日志采集延迟、消息队列积压随便一个环节出问题数据到达Spark的时间就比事件实际发生的时间晚很多。如果没有机制兜底晚到的数据要么被丢弃要么导致之前算好的窗口结果反复变化。Structured Streaming引入的watermark水位线机制本质上是一个数据可晚到多久的约定。你声明withWatermark(event_time, 10 minutes)意思就是告诉引擎所有事件时间晚于当前最大事件时间减去10分钟的数据都会被归入对应的窗口参与计算而更早的、晚到得离谱的数据会被当做过期数据忽略掉对应的窗口也会被关闭并清出内存状态。这里有个细节值得仔细琢磨watermark是建立在事件时间上的不是处理时间。很多人以为10分钟watermark就是允许任务处理延迟10分钟其实不对。它的判断标准是当前已经处理过的数据里最大的event_time减去10分钟作为水位线。举个例子假设上次批次里你看到的最大事件时间是18:00那么水位线就是17:50。如果新来一条数据事件时间是17:45那它会因为早于17:50而被丢弃即使它在系统里只晚到了几秒钟。从代码上看watermark必须和聚合操作放在一起才能生效val result parsedDF .withColumn(event_time, to_timestamp(col(event_time))) .withWatermark(event_time, 10 minutes) .groupBy( window(col(event_time), 5 minutes), col(user_id) ) .count()注意watermark会延迟窗口的关闭时间。如果你想窗口一结束就立刻输出结果watermark设得越长结果输出越晚。比如5分钟窗口配10分钟watermark理论上最早也要到事件时间超过窗口结束10分钟之后那个窗口的结果才会最终确定。这在设计实时报表时非常关键——到底要多实时就需要在水位线的容忍度和输出延迟之间做权衡。3.3 窗口聚合中的输出模式与条件限制窗口聚合搭配Append模式会让很多新手困惑。官方文档的解释是在加了watermark之后Append模式可用于聚合查询因为引擎能够保证一旦一个窗口的结果确定后续不会再更新。所以Append模式输入的是那些已经确定不再变化的窗口结果。Update模式就没有这个限制它可以在每次触发时输出新增或者更新的窗口结果。因此当你的下游系统能够处理更新语义时比如写到可更新的MySQL表用Update模式往往比Append模式更直观。我个人的实际选型经验是如果下游只允许追加写入比如HDFS目录、Kafka优先Append模式搭配足够长的watermark如果下游是支持更新的OLTP数据库或者能覆盖写的数据湖Update模式更舒服调试也方便。4. 状态管理与容错机制4.1 状态存储与状态超时清理有状态的计算比如累计活跃用户数每个用户的会话时长需要引擎在不同批次之间保存中间状态。Structured Streaming内部用状态存储来保存这些中间结果默认情况下状态存储在executor的内存和本地磁盘中由状态管理器定期checkpoint。之前我提到watermark会关闭窗口并清除状态这是状态清理的一种方式。但如果你做的不是窗口聚合而是普通的mapGroupsWithState或者flatMapGroupsWithState自定义状态处理那情况就复杂一些了。以mapGroupsWithState为例你需要自己定义状态的更新逻辑还要定义状态超时时间。官方提供了两种超时策略ProcessingTimeTimeout和EventTimeTimeout。前者以系统当前时间作为超时判断依据适合会话超时这类场景后者以事件时间作为依据和时间窗口语义保持一致。import org.apache.spark.sql.streaming._ val query parsedDF .groupByKey(row row.getAs[String](user_id)) .mapGroupsWithState(GroupStateTimeout.EventTimeTimeout)(updateState) def updateState( userId: String, inputs: Iterator[Row], state: GroupState[UserState] ): UserState { // 自定义状态逻辑 }做自定义状态时我强烈建议你在代码里显式处理state.hasTimedOut的情况否则超时之后旧状态一直滞留内存占用会肉眼可见地往上涨最后任务OOM。这个坑我在做用户行为分析时踩过一次排查了一整天才发现是超时分支没处理。4.2 checkpoint与端到端一致性Structured Streaming最核心的容错机制是checkpoint。你需要给每个query指定一个checkpoint位置通常是HDFS路径或者S3路径。引擎会把读进度offset、状态数据、已提交的输出批次等元数据都写到这个目录。val query parsedDF .writeStream .format(parquet) .outputMode(append) .option(checkpointLocation, /tmp/ss-checkpoint/user_actions) .option(path, /tmp/ss-output/user_actions) .start()很多人会问checkpoint到底保证什么样的语义答案是如果sink具有幂等写入能力Structured Streaming能实现端到端的exactly-once语义。具体来说引擎通过WAL预写日志记录每个批次要处理的offset范围在重启时从WAL恢复确保同一批数据不会因为崩溃而重复读取输出阶段它跟踪已经提交的批次ID并把批次ID传给sink方便下游做幂等去重。这里我想分享一个实际运维中特别容易忽视的点checkpoint路径不要随便换也不要多个query共用同一个checkpointLocation。一旦改动checkpoint路径引擎会认为这是一个全新的查询从头启动之前的offset记录全部失效结果可能从latest开始消费导致数据断层。我有一次因为部署脚本里的路径变量写错导致整个流任务重置白白丢了一个小时的增量数据被下游业务方追着问了半天。4.3 从故障中恢复的正确姿势要是程序真的崩了重启的流程也有讲究。正确的流程是尽可能保持代码逻辑不变只修复你要修复的部分仍然使用同一套checkpoint路径启动。引擎会自动从上次提交的位置继续处理。但如果你改动了查询的输出模式、source配置、聚合逻辑等影响执行计划的配置Spark很可能会抛出一个query plan not compatible之类的异常拒绝启动。这是很多人在开发环境改了代码之后突然发现任务起不来的原因。遇到这种情况成熟的处理方式有两种一是接受数据语义可能变化干脆换一个新的checkpoint路径从头消费二是走真正的升级流程确保新的查询计划能在旧的状态基础上继续不过这一般只支持很小的改动。5. 常见问题排查与性能调优5.1 背压机制与批次间隔怎么设置在Structured Streaming之前的Spark Streaming时代背压需要手动开启之后会自动根据处理耗时调节消费速率。而Structured Streaming相对简单它的触发间隔由trigger控制没有复杂的背压选项。但恰恰是这个简单让很多人忽略了它的存在。默认情况下Structured Streaming会尽可能快地处理也就是前一批次处理完就立刻开始下一批。这样能获得最低的延迟但如果数据源写入速率波动很大会造成处理压力忽高忽低。如果想要固定节奏可以定义ProcessingTime触发器val query parsedDF .writeStream .trigger(Trigger.ProcessingTime(10 seconds)) .format(console) .start()设置一个10秒的processing time trigger之后引擎会每10秒钟检查一次是否有新数据到达并且发起一个新的微批次。这相当于给了下游一个相对稳定的写入节奏。如果你的下游是分析型数据库或者某个不支持高并发写入的存储系统我建议务必设置一个trigger而不是让任务全功率运行否则下游被打爆是早晚的事。5.2 数据倾斜的排查思路Structured Streaming作业最常见的性能杀手不是别的而是数据倾斜尤其在做分组聚合的时候。假设你按user_id分组统计某个头部用户的日志量占全部日志的80%那么处理这个分区的executor会成为瓶颈其他executor空闲等待整个微批次处理时间被拖得很长。排查手段其实和批处理类似。第一看Spark UI中每个task处理的时间分布如果少数task耗时是其他task的几倍甚至几十倍基本可以断定发生了倾斜。第二看executor的内存和GC情况倾斜分区的executor往往内存压力大GC频繁。解决办法也直接。如果是按key分组的聚合可以先加一层随机前缀做二次聚合然后把结果合并。这个方案在批处理里是常规操作流处理里同样适用只不过需要注意加了前缀之后状态管理的key数量会变多状态存储的压力会相应增加。你得在真正动手之前评估一下这个trade-off权衡。对于写sink的时候发生的倾斜比如目标Kafka分区数少于Spark分区数可以开spark.sql.streaming.schemaInference相关的调优参数或者用repartition重新分区保持写出的分区数和下游分区数匹配。5.3 我踩过的几个典型坑第一个坑是Kafka的value值为null。不知道你们有没有遇见过Kafka里的墓碑消息tombstone recordvalue是nullJSON解析还好但一旦你把它当成字符串去split或格式化整个executor就会报错。解决方式是读取之后加一个过滤filter(col(value).isNotNull)。第二个坑是小文件问题。用Structured Streaming写Parquet到HDFS时如果微批次间隔很短、每个批次的数据量又很小生成的Parquet小文件会越积越多之后跑批任务扫描这些文件时目录列表和元数据开销会非常吓人。实践中应对方法有三个层次一是适当增大trigger间隔二是在写之前用coalesce控制输出分区数三是定期跑小文件合并任务。第三个坑是时区问题。事件时间从字符串转Timestamp时如果不显式指定时区Spark默认按照系统时区转化而系统时区通常是UTC。如果你的业务数据事件时间是东八区不处理的话窗口边界会整体偏移8小时统计出来的数据全都不在预期窗口里。排查这类问题时发现数据偏了八个小时十有八九就是时区问题优先检查spark.sql.session.timeZone配置。5.4 一个完整的流任务调试流程建议实操记录调试流任务跟调试批处理任务不太一样因为它是一个持续运行的程序不好直接打断点。我自己的做法是先准备一份历史数据的批量样本也就是把Kafka里最近的几万条数据落成一个临时文件然后用批处理的DataFrame API去模拟流式的处理逻辑。这样做的好处是如果逻辑有问题你在批处理里就能快速发现完全不用等流任务跑起来之后再去翻日志。等批处理逻辑验证通过再用readStreamwriteStream实际跑起来。第一次跑的时候强烈建议直接用console输出先跑个10分钟确认数据格式和结果符合预期然后再把输出源切换到Parquet、Kafka或者下游数据库。很多新人上来就直接往生产sink写结果数据格式不对下游表直接写入失败然后排查了半天才发现是源头数据里多了个隐藏字符。调试时的另一个好习惯是加spark-shell一行一行地执行。比如先读一个批处理DataFrame看看schema再试着调用from_json解析每一行都确认结果最后再把所有逻辑组合成一个copy-paste的脚本。这比一上来就写一整个.scala文件效率高得多因为编译期和运行期的报错信息在shell里更加直观。6. 进阶技巧与学习路线建议6.1 持续处理模式Continuous Processing到底能不能用Structured Streaming在2.3版本中引入了持续处理模式号称能做到毫秒级延迟不再是以微批次为单位而是以连续的方式处理。当时看到这个特性的时候我挺兴奋但真到了生产环境我依然选择微批次模式。原因很简单持续处理模式至今在很多场景中仍是实验性的支持的算子有限、端到端的exactly-once保证不如微批次成熟、状态存储的配合也有不少限制。如果你不需要超低延迟比如业务能接受1秒以上的延迟我建议继续用默认的微批次模式靠谱是第一位的。如果业务确实需要毫秒级延迟那你也该考虑是不是该用更专业的流处理引擎了而不是在Spark生态里硬扛。6.2 和Kafka、数据湖结合的完整架构参考最后说一个我在实际项目中比较常用的技术栈组合。数据源是业务日志通过Kafka接入Structured Streaming负责实时ETL和聚合结果写入Kafka供下游实时服务使用同时备份一份到数据湖用于离线分析。这套架构的好处是实时链路和离线链路共用同一套逻辑只需要把readStream换成read批处理版本几乎不需要改动。这种同一套代码批流一体的能力是我觉得Structured Streaming最有价值的地方。你不用维护两套逻辑也不用手动对账实时数据和离线数据因为它们的计算逻辑本来就是同一份。对于数据团队来说少一个对不齐的口径就少一个扯皮的源头。我也一直在关注社区的最新进展比如Spark 3.x对流处理的持续优化。但不管版本怎么更新核心的模型和思路是稳定的。你花时间吃透这套无界表模型、watermark机制、状态管理和检查点恢复后续版本升级充其量只是让你用上更顺手的工具。最后的最后如果让我给从事流处理开发的同行一个建议那就是别停留在会调API的层面遇到问题的时候多去读一读Spark执行的日志和UI多想想数据从源头到sink的整个生命周期很多所谓的神秘问题其实只是某个环节没有把控好而已。