PySpark Structured Streaming 实时处理 Kafka 数据实战指南

1. 项目概述：为什么一个“Pyspark + Kafka + Structured Streaming”组合值得花三天时间搭通路

如果你正在用 Python 做实时数据处理，又卡在“Kafka 消息一来就丢、Spark Streaming 老是背压崩、窗口计算对不上时间、作业跑两小时就 OOM”这些典型症状上——那这个标题不是技术堆砌，而是一条被踩实的生产级通路。我去年在一家做物联网设备告警分析的团队里，就是靠这套组合，把原本需要 8 小时批处理的设备异常识别，压缩到 90 秒内完成端到端响应。核心不是炫技，而是解决三个真实痛点：消息不丢（exactly-once）、状态可维护（stateful processing）、业务逻辑可读（SQL + DataFrame API 混写）。它不依赖 Flink 的复杂部署，也不用硬啃 Scala，纯 PySpark 写，运维同学能看懂、开发同学能改、测试同学能断言。关键词里 “Structured Streaming” 是关键分水岭——它不是旧版 DStream 那种“微批模拟流”的伪实时，而是基于 Catalyst 优化器和 Tungsten 执行引擎的真正流式执行模型，所有算子（map、filter、window、join）都带语义保证。你不需要成为 Spark 内核专家，但得清楚：Kafka 是你的消息总线，Structured Streaming 是你的流式执行框架，PySpark 是你写业务逻辑的胶水层。适合谁？Python 数据工程师、想从离线转向实时的分析师、中小团队里既要写 ETL 又要盯线上任务的全栈型开发。别被“Pipeline”这个词吓住——它本质就是一条数据流动的高速公路：Kafka 生产端发车 → Spark 流式应用接单 → 经过清洗/聚合/关联 → 最终落库或触发告警。下面拆解的每一步，我都配了生产环境实测过的参数、命令和避坑点，不是文档翻译，是凌晨三点调通 checkpoint 后记下的笔记。

2. 整体架构设计与方案选型逻辑：为什么不用 Flink？为什么坚持用 Kafka？为什么 Structured Streaming 不是“过渡方案”

2.1 架构全景图：三层收敛，拒绝过度设计

我们最终落地的架构只有三层，没有网关、没有中间缓存、没有双写冗余：

• 接入层（Kafka）：仅承担消息缓冲与解耦，Topic 设计严格按业务域划分（如iot.device.raw、iot.device.enriched），分区数按峰值吞吐预估（公式见后），不启用压缩（CPU 换延迟更可控）；
• 计算层（PySpark Structured Streaming）：单个 Spark Structured Streaming 应用，消费 Kafka Topic，执行 stateful 窗口聚合，结果写入 Delta Lake（非 Hive）；
• 服务层（Delta Lake + REST API）：Delta 表作为唯一事实源，通过 Delta Sharing 或轻量 Flask API 对外提供查询，避免 Spark 直连下游 DB 的连接风暴。

这个设计砍掉了所有“看起来高大上但增加故障点”的组件。比如没上 Redis 做实时指标缓存——因为 Delta Lake 的 Z-ordering + data skipping 在 10 亿级小文件场景下，SELECT COUNT(*) WHERE ts > '2024-06-01'查询比 RedisINCRBY还快 200ms；也没用 Kafka Connect 做 Sink——因为 Structured Streaming 自带foreachBatch，能精准控制事务边界，避免 Connect 的 at-least-once 语义导致重复写入。

2.2 为什么不是 Flink？——成本、生态与团队能力的三角权衡

Flink 确实更“原生流式”，但我们在技术选型会上用三组数据否决了它：

• 人力成本：团队 5 名 Python 工程师，0 名 Java/Scala 开发。Flink SQL 虽好，但遇到OVER WINDOW性能瓶颈时，必须下钻到 ProcessFunction 写 Java。而 PySpark 的pandas_udf+groupby.window()组合，调试周期从 3 天缩到 4 小时；
• 运维成本：Flink on YARN 需要独立的 JobManager HA 配置、TaskManager 内存精细化调优、Checkpoint 存储路径权限管理。而 Spark Streaming 应用直接复用现有 YARN 集群，spark-submit一条命令启停，日志统一走 ELK；
• 生态粘性：我们已有 200+ 个离线 PySpark 任务跑在 Airflow 上。Streaming 任务复用同一套调度、监控、告警体系，spark.sql.adaptive.enabled=true自适应查询优化还能让流批 SQL 共享同一套统计信息。

提示：Flink 在超低延迟（<100ms）或复杂事件处理（CEP）场景仍是首选。但如果你的 SLA 是“秒级响应”，且团队主语言是 Python，Structured Streaming 是更务实的选择。

2.3 为什么 Kafka 是不可替代的消息中间件？

有人问：“用 Pulsar 不是更云原生？”、“RabbitMQ 不是更轻量？”——我们做过压测对比（10 万设备并发上报，每秒 5000 条 JSON）：

Kafka 的胜出点在于“足够好且足够稳”：ISR（In-Sync Replicas）机制确保 Leader 挂掉时，Follower 能无损接管；kafka-python 客户端经过十年迭代，enable.auto.commit=False+commit_async()组合能精确控制 offset 提交时机；更重要的是，它的分区模型天然匹配 Spark 的并行度——每个 Kafka 分区对应一个 Spark 任务，彻底规避了“热点分区导致 task skew”。

2.4 Structured Streaming 为什么不是“过渡方案”？——从 DStream 到 Structured 的范式升级

很多老司机还在用 DStream，觉得“够用”。但 DStream 的本质缺陷无法绕过：

• 微批陷阱（Micro-batch Illusion）：DStream 把流切成 1s/2s 的小批，窗口计算实际是“对一批数据做聚合”，而非“对连续时间轴上的事件做滑动”。这导致：当 Kafka 消息乱序（如设备时钟漂移），DStream 无法按事件时间（event-time）正确归窗；
• 状态管理黑盒：DStream 的updateStateByKey状态存储在 Driver 内存，扩容即失效；而 Structured Streaming 的mapGroupsWithState状态持久化到 RocksDB + HDFS，支持横向扩展；
• 语义保障缺失：DStream 的 exactly-once 依赖外部系统（如 Kafka + HDFS 事务），而 Structured Streaming 原生支持checkpointLocation+kafka.offsetsByTimestamp实现端到端 exactly-once。

我们曾用 DStream 做 10 分钟滚动窗口的设备在线率统计，因网络抖动导致某批消息延迟 3 秒到达，结果该批数据被错误地计入下一个窗口，误差率达 17%。切换到 Structured Streaming 后，开启eventTime+watermark，问题消失。

3. 核心细节解析与实操要点：Kafka Topic 设计、Spark 参数调优、状态管理实战

3.1 Kafka Topic 设计：分区数不是拍脑袋，而是算出来的

分区数决定并行度上限，也影响数据倾斜风险。我们用这个公式确定iot.device.rawTopic 的初始分区数：

分区数 = max(ceil(峰值吞吐量 / 单分区吞吐量), ceil(消费者并发数))

其中：

• 峰值吞吐量：根据设备上报频率 × 设备总数 × 单条消息大小估算。例如：10 万台设备，每 30 秒上报 1 条 500B 消息 → 峰值 QPS = 100000 / 30 ≈ 3333 条/秒 → 峰值吞吐 = 3333 × 0.5KB ≈ 1.6MB/s；
• 单分区吞吐量：Kafka 官方建议单分区稳定吞吐 ≤ 10MB/s（SSD 磁盘），保守取 5MB/s；
• 消费者并发数：Spark Streaming 应用的spark.sql.streaming.kafka.maxOffsetsPerTrigger设置为 10000，每个 micro-batch 处理约 10000 条，按 3333 QPS 计算，需约 3 个并发 task。

代入公式：max(ceil(1.6 / 5), ceil(3)) = max(1, 3) = 3。但这是理论最小值，我们实际设为12 分区——原因有三：

1. 预留 300% 扩容空间（设备数可能翻倍）；
2. 规避单节点磁盘 IO 瓶颈（12 分区可均匀分布到 3 台 broker）；
3. Spark 并行度需整除分区数（12 能被 2/3/4/6 整除，适配不同 executor 数量）。

注意：分区数一旦创建不能减少，只能增加。增加后需手动 reassign partitions，否则新分区无数据。我们用kafka-reassign-partitions.sh脚本 + JSON 配置文件完成迁移，全程 5 分钟，零丢失。

Topic 创建命令（含关键参数）：

bin/kafka-topics.sh --create \ --bootstrap-server kafka1:9092,kafka2:9092,kafka3:9092 \ --topic iot.device.raw \ --partitions 12 \ --replication-factor 3 \ --config retention.ms=604800000 \ # 保留 7 天，平衡存储与重放需求 --config segment.bytes=1073741824 \ # 1GB 段大小，减少小文件 --config min.insync.replicas=2 \ # 至少 2 副本同步才认为写成功 --config cleanup.policy=compact \ # 启用 log compaction，支持 key-based dedup --config compression.type=none

3.2 Spark Structured Streaming 核心参数调优：内存、并行度、背压的黄金配比

一个常被忽略的事实：Structured Streaming 的性能 70% 取决于 Spark 通用参数，而非流式专属参数。我们在 16 核 64GB 内存的 YARN 集群上，摸索出这套生产级配置：

Executor 资源分配（关键！）

# spark-submit 参数 --num-executors 6 \ --executor-cores 4 \ --executor-memory 12G \ --driver-memory 4G \ --conf spark.sql.adaptive.enabled=true \ --conf spark.sql.adaptive.coalescePartitions.enabled=true \ --conf spark.sql.adaptive.skewJoin.enabled=true

• 为什么是 6 个 executor？Kafka 有 12 个分区，每个 executor 分配 2 个分区（spark.sql.streaming.kafka.maxOffsetsPerTrigger默认按分区均衡分配），避免单 executor 过载；
• 为什么 executor-memory=12G？剩余 4G 给 OS 和 JVM Overhead（spark.executor.memoryOverhead=4096），防止 container 被 YARN kill；
• 自适应查询（AQE）必开：它能动态合并小 partition（避免 1000+ 个小 task 拖慢整体）、自动处理数据倾斜（无需手动salting）。

流式专属参数（直接影响 exactly-once）

df = spark \ .readStream \ .format("kafka") \ .option("kafka.bootstrap.servers", "kafka1:9092,kafka2:9092,kafka3:9092") \ .option("subscribe", "iot.device.raw") \ .option("startingOffsets", "earliest") \ # 首次启动从头消费 .option("endingOffsets", "latest") \ # 测试时可设为 latest 快速验证 .option("failOnDataLoss", "false") \ # 防止 Kafka 删除旧日志导致 job crash .option("kafka.group.id", "spark-streaming-iot") \ # 固定 group.id，保证 offset 可追溯 .option("kafka.request.timeout.ms", "60000") \ .option("kafka.session.timeout.ms", "30000") \ .load()

• kafka.group.id必须固定：Structured Streaming 依赖它从 Kafka 的__consumer_offsets主题读取初始 offset，若每次随机生成，将丢失历史位置；
• failOnDataLoss=false是血泪教训：某次 Kafka 磁盘满，自动清理了 3 天前日志，若设为 true，job 直接失败；设为 false 后，Spark 自动跳过丢失 offset，从最早可用位置继续。

Checkpoint 机制：状态与 offset 的双重保险

query = df \ .selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)") \ .select(from_json(col("value"), schema).alias("data")) \ .select("data.*") \ .withColumn("event_time", col("timestamp").cast("timestamp")) \ .withWatermark("event_time", "10 minutes") \ # 允许 10 分钟乱序 .groupBy( window(col("event_time"), "10 minutes", "5 minutes"), # 10 分钟窗口，5 分钟滑动 col("device_id") ) \ .agg( count("*").alias("msg_count"), avg("temperature").alias("avg_temp") ) \ .writeStream \ .format("delta") \ .outputMode("Append") \ # 窗口聚合用 Append，避免重复写入 .option("checkpointLocation", "hdfs://namenode:8020/checkpoints/iot-device-window") \ .start()

• checkpointLocation必须指向高可用存储（HDFS/S3），它存储三类关键数据：
  1. Kafka offset（每次 micro-batch 结束后提交到 Kafka）；
  2. Streaming 状态（如mapGroupsWithState的 RocksDB 快照）；
  3. 逻辑计划（用于故障恢复时重建 query plan）。
• 若 checkpoint 路径损坏，job 无法恢复——我们每天凌晨用hdfs dfs -du -s监控其大小，超过 5GB 触发告警（正常应 < 500MB）。

3.3 Stateful Processing 实战：用mapGroupsWithState替代reduce做设备心跳检测

窗口聚合解决不了“设备是否在线”这类状态型问题。例如：设备每 30 秒发一次心跳，若连续 3 分钟无心跳，则标记为离线。这需要跨多个 micro-batch 维护设备状态。mapGroupsWithState是 Structured Streaming 提供的低阶 API，比flatMapGroupsWithState更轻量（不支持 timeout）。

from pyspark.sql.functions import * from pyspark.sql.types import * # 定义状态 Schema state_schema = StructType([ StructField("last_heartbeat", TimestampType(), True), StructField("offline_since", TimestampType(), True), StructField("is_online", BooleanType(), True) ]) def update_device_state(key, values, state): """ key: device_id values: 当前 batch 中该 device_id 的所有记录（list of Row） state: 当前状态（Row 或 None） """ from datetime import timedelta # 初始化状态 if state is None: state = Row(last_heartbeat=None, offline_since=None, is_online=True) # 获取当前 batch 中最新心跳时间 current_max_ts = max([row.event_time for row in values]) if values else None # 更新状态逻辑 if current_max_ts: # 有新心跳：重置在线状态 new_state = Row( last_heartbeat=current_max_ts, offline_since=None, is_online=True ) yield Row(device_id=key, event_time=current_max_ts, status="online", state_updated=True) else: # 无新心跳：检查是否超时 if state.last_heartbeat and (current_max_ts is None or (current_max_ts - state.last_heartbeat) > timedelta(minutes=3)): # 超过 3 分钟未心跳 if state.is_online: # 刚离线，记录离线时间 offline_time = state.last_heartbeat + timedelta(minutes=3) new_state = Row( last_heartbeat=state.last_heartbeat, offline_since=offline_time, is_online=False ) yield Row(device_id=key, event_time=offline_time, status="offline", state_updated=True) else: # 已离线，保持状态 new_state = state else: # 未超时，保持在线 new_state = Row( last_heartbeat=state.last_heartbeat, offline_since=state.offline_since, is_online=True ) # 更新状态（必须调用） state.update(new_state) # 注册 UDF update_func = pandas_udf(update_device_state, returnType=...) # 应用到流 stateful_df = df \ .withColumn("event_time", col("timestamp").cast("timestamp")) \ .withWatermark("event_time", "10 minutes") \ .groupBy("device_id") \ .applyInPandas(update_func, "device_id STRING, event_time TIMESTAMP, status STRING, state_updated BOOLEAN")

实操心得：mapGroupsWithState的状态默认存在 executor 内存，但生产环境必须配置spark.sql.streaming.stateStore.providerClass=org.apache.spark.sql.execution.streaming.state.RocksDBStateStoreProvider，将状态持久化到 RocksDB（本地磁盘）+ HDFS（快照备份）。否则 executor 挂掉，状态全丢。

4. 实操过程与核心环节实现：从零搭建端到端 Pipeline 的完整步骤

4.1 环境准备与依赖安装：避开 Python 版本与 Spark 版本的兼容雷区

我们锁定Spark 3.4.1 + Python 3.9 + Kafka 3.4.0组合，这是目前最稳定的三角关系。Spark 3.5+ 对 Python 3.11 支持不完善，Kafka 3.5+ 的 SASL 认证在 PySpark 中偶发 handshake timeout。

Step 1：安装 Kafka Python 客户端

# 必须指定版本！kafka-python 2.0+ 与 Spark 3.4 兼容性更好 pip install kafka-python==2.0.2 # 验证：能连接 Kafka 并列出 topic python -c " from kafka import KafkaClient client = KafkaClient(bootstrap_servers='kafka1:9092') print(client.cluster.topics()) "

Step 2：下载 Spark Kafka 包（关键！）Structured Streaming 的 Kafka source 不在 Spark core 中，需额外加载：

# 下载 spark-sql-kafka-0-10_2.12-3.4.1.jar # 地址：https://repo1.maven.org/maven2/org/apache/spark/spark-sql-kafka-0-10_2.12/3.4.1/ # 放入 $SPARK_HOME/jars/ 目录 ls $SPARK_HOME/jars/spark-sql-kafka-0-10_2.12-3.4.1.jar

注意：jar 包名中的_2.12表示 Scala 版本，必须与 Spark 编译的 Scala 版本一致（Spark 3.4 默认用 Scala 2.12）。若用 Scala 2.13 编译的 Spark，需下载_2.13版本，否则ClassNotFoundException。

Step 3：配置 Spark Session（含 Kafka 认证）

from pyspark.sql import SparkSession from pyspark.sql.types import * spark = SparkSession.builder \ .appName("iot-device-streaming") \ .config("spark.sql.adaptive.enabled", "true") \ .config("spark.sql.adaptive.coalescePartitions.enabled", "true") \ .config("spark.sql.adaptive.skewJoin.enabled", "true") \ .config("spark.sql.streaming.checkpointLocation", "hdfs://namenode:8020/checkpoints/iot-device") \ .config("spark.jars", "/opt/spark/jars/spark-sql-kafka-0-10_2.12-3.4.1.jar") \ .getOrCreate() # 若 Kafka 启用 SASL/PLAIN 认证（生产环境必须） spark.conf.set("spark.sql.kafka.sasl.jaas.config", "org.apache.kafka.common.security.plain.PlainLoginModule required username='kafka-user' password='kafka-pass';") spark.conf.set("spark.sql.kafka.sasl.mechanism", "PLAIN") spark.conf.set("spark.sql.kafka.security.protocol", "SASL_PLAINTEXT")

4.2 数据接入与解析：从 Kafka Raw Bytes 到 Structured DataFrame

Kafka 的value是字节数组，必须反序列化。我们采用JSON Schema + Avro混合策略：设备原始数据用 JSON（开发快），关键指标用 Avro（节省带宽）。

JSON 解析（推荐新手起步）

# 定义设备数据 Schema（严格匹配 Kafka 消息结构） device_schema = StructType([ StructField("device_id", StringType(), False), StructField("timestamp", LongType(), False), # Unix timestamp in ms StructField("temperature", DoubleType(), True), StructField("humidity", DoubleType(), True), StructField("battery_level", IntegerType(), True), StructField("status", StringType(), True) # "online", "offline", "error" ]) # 读取 Kafka 并解析 kafka_df = spark \ .readStream \ .format("kafka") \ .option("kafka.bootstrap.servers", "kafka1:9092,kafka2:9092,kafka3:9092") \ .option("subscribe", "iot.device.raw") \ .option("startingOffsets", "earliest") \ .option("kafka.group.id", "spark-streaming-iot") \ .load() # 解析 value 字段（key 通常为 device_id，可选） parsed_df = kafka_df \ .select( col("key").cast("string").alias("device_id"), from_json(col("value").cast("string"), device_schema).alias("data") ) \ .select("device_id", "data.*") \ .withColumn("event_time", from_unixtime(col("timestamp") / 1000).cast("timestamp")) \ .withWatermark("event_time", "10 minutes") # 输出 schema 验证 parsed_df.printSchema() # root # |-- device_id: string (nullable = true) # |-- timestamp: long (nullable = true) # |-- temperature: double (nullable = true) # |-- humidity: double (nullable = true) # |-- battery_level: integer (nullable = true) # |-- status: string (nullable = true) # |-- event_time: timestamp (nullable = true)

Avro 解析（生产环境推荐）

# 需先安装 avro-python3 pip install avro-python3 # 使用 Confluent Schema Registry 获取 Avro Schema from confluent_kafka.schema_registry import SchemaRegistryClient from confluent_kafka.schema_registry.avro import AvroDeserializer schema_registry_conf = {'url': "http://schema-registry:8081"} schema_registry_client = SchemaRegistryClient(schema_registry_conf) avro_deserializer = AvroDeserializer(schema_registry_client, device_schema) # 在 Spark UDF 中调用（略，因涉及 Java UDF 注册，此处省略详细代码） # 核心优势：Avro 二进制序列化比 JSON 小 60%，网络传输更快

4.3 核心业务逻辑实现：设备异常检测的 3 层过滤流水线

我们构建了一个“轻量级但可解释”的异常检测 Pipeline，不依赖 ML 模型，用规则引擎保障可维护性：

Layer 1：基础数据质量过滤（防脏数据污染后续计算）

# 过滤明显异常值（温度 > 100°C 或 < -50°C） quality_filtered_df = parsed_df \ .filter( (col("temperature").between(-50.0, 100.0)) & (col("humidity").between(0.0, 100.0)) & (col("battery_level").between(0, 100)) ) \ .filter(col("device_id").isNotNull()) \ .filter(col("timestamp").isNotNull()) # 添加数据质量标记 quality_df = quality_filtered_df \ .withColumn("data_quality_score", when(col("temperature").between(15.0, 35.0), 1.0) .when(col("temperature").between(0.0, 50.0), 0.8) .otherwise(0.5)) \ .withColumn("is_quality_pass", col("data_quality_score") >= 0.8)

Layer 2：设备行为模式识别（滑动窗口统计）

# 计算设备最近 5 分钟的平均温度、标准差、上报频率 windowed_stats = quality_df \ .withWatermark("event_time", "10 minutes") \ .groupBy( window(col("event_time"), "5 minutes", "1 minute"), # 5 分钟窗口，1 分钟滑动 col("device_id") ) \ .agg( avg("temperature").alias("avg_temp_5m"), stddev("temperature").alias("std_temp_5m"), count("*").alias("msg_count_5m"), max("event_time").alias("last_report_time") ) \ .select("window.*", "device_id", "avg_temp_5m", "std_temp_5m", "msg_count_5m", "last_report_time") # 关联原始数据，丰富上下文 enriched_df = windowed_stats \ .join( quality_df.select("device_id", "event_time", "temperature", "humidity", "battery_level"), ["device_id"], "left" ) \ .filter(col("event_time") >= col("window.start")) \ .filter(col("event_time") <= col("window.end"))

Layer 3：业务规则引擎（触发告警）

# 定义异常规则（可配置化，存入数据库或配置中心） abnormal_rules = [ # 规则1：温度突变（标准差 > 5°C 且均值偏离历史基线 10°C） (col("std_temp_5m") > 5.0) & (abs(col("avg_temp_5m") - 25.0) > 10.0), # 规则2：上报频率骤降（5 分钟内消息数 < 5 条，正常应为 10 条） col("msg_count_5m") < 5, # 规则3：电池电量低于阈值且持续下降 (col("battery_level") < 20) & (col("battery_level") < lag("battery_level", 1).over(Window.partitionBy("device_id").orderBy("event_time"))) ] # 应用规则 alert_df = enriched_df \ .withColumn("abnormal_flag", when(abnormal_rules[0], "temp_spike") .when(abnormal_rules[1], "low_frequency") .when(abnormal_rules[2], "low_battery") .otherwise("normal")) \ .filter(col("abnormal_flag") != "normal") \ .withColumn("alert_id", expr("uuid()")) \ .withColumn("alert_time", current_timestamp()) # 输出告警到 Kafka（供下游通知服务消费） alert_query = alert_df \ .select( col("alert_id").cast("string").alias("key"), to_json(struct( col("device_id"), col("alert_time"), col("abnormal_flag"), col("avg_temp_5m"), col("msg_count_5m") )).cast("string").alias("value") ) \ .writeStream \ .format("kafka") \ .option("kafka.bootstrap.servers", "kafka1:9092,kafka2:9092,kafka3:9092") \ .option("topic", "iot.alerts") \ .option("checkpointLocation", "hdfs://namenode:8020/checkpoints/iot-alerts") \ .outputMode("Append") \ .start()

4.4 结果输出与监控：Delta Lake 写入与健康度仪表盘

Delta Lake 写入（替代 Hive，支持 ACID 与 Time Travel）

# 写入 Delta 表（自动创建表结构） delta_query = enriched_df \ .writeStream \ .format("delta") \ .outputMode("Append") \ .option("checkpointLocation", "hdfs://namenode:8020/checkpoints/iot-enriched") \ .table("iot_device_enriched") # 自动在 Spark Catalog 中注册 # 启用 Z-ordering 加速查询（按高频查询字段） spark.sql("OPTIMIZE iot_device_enriched ZORDER BY (device_id, event_time)") # 查询 1 小时内设备在线率（毫秒级响应） spark.sql(""" SELECT COUNT(DISTINCT CASE WHEN status = 'online' THEN device_id END) * 100.0 / COUNT(DISTINCT device_id) as online_rate FROM iot_device_enriched WHERE event_time >= current_timestamp() - interval 1 hours """).show()

健康度监控（用 Spark UI + 自定义 Metrics）

# 在 StreamingQueryListener 中埋点 class StreamingMetricsListener(StreamingQueryListener): def onQueryStarted(self, event): print(f"Query started: {event.id}") def onQueryProgress(self, event): progress = event.progress # 上报关键指标到 Prometheus metrics.gauge('streaming_input_rows', progress.inputRowsPerSecond) metrics.gauge('streaming_processed_rows', progress.processedRowsPerSecond) metrics.gauge('streaming_batch_duration_ms', progress.durationMs['addBatch']) metrics.gauge('streaming_state_size_bytes', progress.stateOperators[0].numRowsTotal if progress.stateOperators else 0) # 注册监听器 spark.streams.addListener(StreamingMetricsListener())

5. 常见问题与排查技巧实录：那些文档里不会写的血泪经验

5.1 Kafka Offset 提交失败：CommitFailedException的根因与解法

现象：日志中频繁出现ERROR KafkaSource: Failed to commit offsets，随后 job 挂起或重启。

根因分析：这不是 Kafka 问题，而是 Spark 的kafka.group.id配置冲突。当多个 Streaming job 使用相同group.id，Kafka 认为它们是同一个 consumer group，会触发 rebalance。若一个 job 正在 commit offset，另一个 job 同时 rebalance，就会抛CommitFailedException。

排查步骤：

1. 查看 Kafka 消费者组状态：bin/kafka-consumer-groups.sh --bootstrap-server kafka1:9092 --group spark-streaming-iot --describe
2. 检查MEMBERS列是否有多个 client id（格式为spark-kafka-source-xxx）；
3. 检查LAG列是否持续增长（说明 offset 未提交成功）。

解决方案：

• 强制隔离：为每个 job 设置唯一kafka.group.id，如spark-streaming-iot-raw、spark-streaming-iot-alert；
• 延长 session timeout：在 Spark conf 中添加.option("kafka.session.timeout.ms", "60000")，避免网络抖动触发误 rebalance；
• 禁用 auto-commit：.option("kafka.enable.auto.commit", "false")，完全由 Spark 控制 commit 时机。

实操心得：我们曾因共用 group.id 导致 3 个 job 相互干扰，延迟从 2 秒飙升到 45 秒。改用唯一 group.id 后，P99 延迟稳定在 1.8 秒。

5.2 State Store 磁盘爆满：RocksDB 占用 200GB 且不释放

现象：/var/data/rocksdb/目录持续增长，df -h显示磁盘使用率 > 95%，Spark executor 频繁 OOM。

根因：RocksDB 的block_cache和write_buffer默认使用堆外内存，但若配置不当，会大量写入本地磁盘。更致命的是，Structured Streaming 的stateStore不会自动清理过期状态（即使设置了 watermark），需手动触发clearExpiredState。

解法：

1. 限制 RocksDB 本地磁盘使用：

spark.conf.set("spark.sql.streaming.stateStore.providerClass", "org.apache.spark.sql.execution.streaming.state.RocksDBStateStoreProvider") spark.conf.set("spark.sql.streaming.stateStore.rocksdb.options", "write_buffer_size=64MB;max_write_buffer_number=4;max_background_compactions=2")

2. 强制清理过期状态（在 foreachBatch 中）：

def process_batch