做实时数据处理这几年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 groupIdorg.apache.spark/groupId artifactIdspark-streaming_2.12/artifactId version3.3.2/version /dependency dependency groupIdorg.apache.spark/groupId artifactIdspark-streaming-kafka-0-10_2.12/artifactId version3.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(sAPI$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作业的性能很大程度上取决于并行度和内存配置。我整理了一份我常用的参数清单大家可以当模板用。参数推荐配置说明spark.streaming.kafka.maxRatePerPartition1000~10000限制每个Kafka分区每秒拉取的最大记录数防止批次堆积spark.streaming.backpressure.enabledtrue开启背压机制让系统自动调节摄入速率spark.streaming.backpressure.initialRate5000背压生效前初始阶段的摄入速率spark.sql.shuffle.partitions200~500控制shuffle并行度流任务建议按数据量调大spark.streaming.stopGracefullyOnShutdowntrue优雅停止先把积压批次处理完再退出spark.executor.memory4G以上executor内存堆内内存主要用来缓存批次数据spark.memory.offHeap.enabledtrue配合offHeap.size使用减少GC压力spark.memory.offHeap.size2Goffheap内存上限给内存占用大的场景留余地如果你用的是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.providerClassorg.apache.spark.sql.execution.streaming.state.RocksDBStateStoreProvider要注意的是RocksDB本身就是一个小型嵌入式数据库它需要一些磁盘IO。如果你的executor是部署在本地磁盘上注意预留足够的空间。我见过一个集群因为RocksDB的本地目录被打满导致状态写入失败任务不断重启。最后在Spark配置里给stateStore目录指定了一个容量更大的路径才解决问题。5.3 批次积压的排查思路批次积压是最常见的生产事故。表现就是Spark UI里的Scheduling Delay越来越大批处理时间逐步拉长最后整个作业处于“半死不活”的状态。我的排查步骤是先看是不是输入速率激增。是Kafka消息量涨了还是上游重试导致重复生产。前者需要加资源后者要上下游协调。再看是不是某个stage执行特别慢。点开Spark UI看每个stage的耗时分布。如果某个stage的task计算时间差异巨大说明有数据倾斜需要加salting或者重新分区。看GC情况。如果executor的GC时间占比很高说明堆内内存紧张考虑增加executor内存、降低每批数据量或者开启offheap。看是否有背压生效。如果maxRatePerPartition没有正确生效你会看到输入速率一直保持很高的值这时候要检查是不是启动参数没有读到配置文件。这个排查顺序基本能把80%的问题定位出来。剩下20%往往是下游依赖的问题比如写Kafka时目标topic的partition数不够或者写入HDFS时NameNode压力大。6. 常见问题与排查技巧实录6.1 几个典型的“翻车”现场我把自己以前遇到过的、以及帮别人排查过的几类高频问题整理成了一张表问题现象原因解决方案作业重启后Kafka offset丢失重复消费checkpoint目录丢失或更换了固定checkpoint目录定期备份元数据窗口聚合结果一直为空事件时间字段类型错误或未设置watermark用正确的时间戳类型显式设置watermark输出到Kafka时报value不能为null聚合结果没有命名字段value用selectExpr(CAST(... AS STRING) AS value)DStream的state丢失checkpoint未配置或配置了不稳定的目录检查checkpoint路径权限配置到HDFS背压不生效spark.streaming.backpressure.enabled未开启显式设置true任务卡住不消费Kafka分区数小于executor并发度增加Kafka topic分区数或调整接受策略RocksDB目录空间不足本地临时目录太小设置spark.sql.streaming.stateStore.rocksdb.directory到更大路径多个作业共用同一个checkpointLocation作业间共享了checkpoint每个作业独立checkpoint目录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不是银弹但从“批”到“微批”再到“真正的流”的演进路线里它是绕不开的一块基石。搞懂它、用好它你在实时数据处理上的底盘就算稳了。