claude-skills Spark Engineer 参考手册解读Apache Spark Structured Streaming 流处理实战模式全解析【免费下载链接】claude-skills67 Specialized Skills for Full-Stack Developers. Transform Claude Code into your expert pair programmer.项目地址: https://gitcode.com/GitHub_Trending/claud/claude-skillsStructured Streaming 是 Apache Spark 面向连续数据流Kafka、文件、Socket构建可扩展、容错、端到端精确一次exactly-once语义流处理应用的核心 API。本文以 claude-skills 项目中 spark-engineer 技能的 流处理参考文档 为骨架系统讲解从流数据源读取、输出模式选择、Watermark 与事件时间处理、窗口与有状态计算到流式 Join、Sink、触发器、监控与性能调优的完整实战路径并结合仓库内源码级参考资料给出可验证的配置依据。读完本文你将掌握一套可直接落地的 PySpark/Scala Structured Streaming 作业设计模式与反模式清单。Structured Streaming 概览与适用场景Structured Streaming 的核心思想是把无限流当作一张持续追加的“无限表”每个触发间隔trigger interval到达的新数据形成一个“微批次”micro-batch引擎以 DataFrame/SQL 的方式增量处理这些新行。它复用了 Spark SQL 的 Catalyst 优化器与 Tungsten 执行引擎因此流式查询与批式查询在 API 层面高度统一。何时使用 Structured Streaming参考文档列出的适用场景非常明确处理连续数据流Kafka、文件、Socket 等来源需要端到端精确一次exactly-once处理保证实时分析与实时仪表盘事件驱动架构从流式数据源做增量 ETL。何时应改用其他方案批处理已能满足需求时优先用批处理复杂度更低需要亚秒级延迟时考虑 FlinkStructured Streaming 的微批次模型天然带来调度开销事件处理非常简单时Kafka Streams 可能就足够。仓库佐证在 SKILL.md 中spark-engineer 技能将references/streaming-patterns.md定义为“Structured Streaming、watermarks、stateful operations、sinks”专题的按需加载参考并把它列为 Spark 作业开发五大参考主题之一其余为 DataFrame/SQL、RDD、分区缓存、性能调优可见流处理模式在本技能知识体系中的核心地位。读取流式数据源Kafka、文件与 Rate一切流式作业都从spark.readStream开始。参考文档覆盖了三种最常用的输入源。Kafka SourceKafka 是流处理最主流的数据源。读取时通过kafka.前缀的 option 透传 Kafka 客户端配置并可用startingOffsets控制消费起点、maxOffsetsPerTrigger控制单批拉取上限背压机制# Read from Kafka df spark.readStream \ .format(kafka) \ .option(kafka.bootstrap.servers, broker1:9092,broker2:9092) \ .option(subscribe, topic1,topic2) \ .option(startingOffsets, latest) \ .option(maxOffsetsPerTrigger, 100000) \ .option(kafka.security.protocol, SASL_SSL) \ .option(kafka.sasl.mechanism, PLAIN) \ .load()Kafka 源输出的固定字段中key与value均为字节数组binarypartition、offset、timestamp分别对应分区号、偏移量与写入时间。因此第一步几乎总是做反序列化——例如把 value 中的 JSON 解析成结构化 schemafrom pyspark.sql import functions as F from pyspark.sql.types import StructType, StructField, StringType, TimestampType, DoubleType schema StructType([ StructField(event_id, StringType()), StructField(user_id, StringType()), StructField(event_time, TimestampType()), StructField(amount, DoubleType()) ]) parsed_df df.select( F.col(key).cast(string).alias(kafka_key), F.from_json(F.col(value).cast(string), schema).alias(data), F.col(timestamp).alias(kafka_timestamp), F.col(partition), F.col(offset) ).select(kafka_key, data.*, kafka_timestamp, partition, offset)对应的 Scala 写法// Scala Kafka source val df spark.readStream .format(kafka) .option(kafka.bootstrap.servers, broker1:9092,broker2:9092) .option(subscribe, topic1) .option(startingOffsets, latest) .load() val parsed df.select( col(key).cast(string), from_json(col(value).cast(string), schema).as(data) ).select(key, data.*)实践要点F.from_json配合显式StructType是最可控的反序列化方式。若消息结构不固定可结合 spark-sql-dataframes.md 中“生产环境必须显式定义 schema”的规范——避免依赖 schema 推断带来的全量扫描与类型漂移问题。File Source文件自动发现当文件以约定前缀/目录落盘时如日志收集器持续产出File Source 会自动发现并读取新增文件天然具备断点续读能力# Read new files as they arrive df spark.readStream \ .format(parquet) \ .schema(my_schema) \ .option(path, s3://bucket/incoming/) \ .option(maxFilesPerTrigger, 100) \ .load() # For JSON files df spark.readStream \ .format(json) \ .schema(my_schema) \ .option(path, s3://bucket/incoming/) \ .load() # CSV with header df spark.readStream \ .format(csv) \ .schema(my_schema) \ .option(path, s3://bucket/incoming/) \ .option(header, true) \ .load()maxFilesPerTrigger控制每个批次最多处理多少新文件是文件源的背压手段。Rate Source测试专用Rate Source 按指定速率持续生成 (timestamp, value) 测试数据其中 value 为递增 long用于验证流式逻辑与基准测试无需真实消息系统# Generate test data at specified rate df spark.readStream \ .format(rate) \ .option(rowsPerSecond, 1000) \ .option(numPartitions, 10) \ .load() # Columns: timestamp, value (incrementing long)仓库佐证关于并行度的参数含义可对照 partitioning-caching.md 中“每个 CPU 核心 24 个分区、目标分区大小 128MB256MB”的经验法则——numPartitions直接决定流任务并发度应与执行资源匹配。输出模式Append / Update / Complete输出模式决定“每个触发周期向 Sink 写入什么”是流式查询结果语义的核心开关。Append Mode默认只写自上次触发以来新增的行。适用于无聚合的场景或配合 Watermark 使用的事件时间窗口聚合# Only new rows added since last trigger # Use when: No aggregations, or windowed aggregations with watermark query df.writeStream \ .outputMode(append) \ .format(parquet) \ .option(path, s3://bucket/output/) \ .option(checkpointLocation, s3://bucket/checkpoints/) \ .start()Update Mode只写自上次触发以来发生变化的行聚合结果中 key 对应值更新了才输出。适用于流式聚合、需要增量结果更新的场景也是控制台观察流式聚合的首选# Only rows that changed since last trigger # Use when: Aggregations, want incremental updates query df.groupBy(user_id).count() \ .writeStream \ .outputMode(update) \ .format(console) \ .start()Complete Mode每个触发周期输出整个结果表。适合需要完整聚合状态的仪表盘但状态量大时开销极高# Entire result table every trigger # Use when: Need full aggregation result each time # Warning: Can be expensive for large state query df.groupBy(user_id).count() \ .writeStream \ .outputMode(complete) \ .format(console) \ .start()模式选择速查表Use CaseOutput ModeNotesETL to filesappend默认最高效Windowed aggregationsappend需配合 WatermarkRunning counts/sumsupdate增量输出Dashboards needing full statecomplete开销大Deduplicationappend配合 dropDuplicatesWatermark 与事件时间流数据乱序out-of-order是常态。Watermark 定义了“迟到数据的最晚容忍线”其语义为max(事件时间) − watermark 时长早于该线的事件将被丢弃同时引擎得以清理过期状态有界内存、在恰当时机发出聚合结果。设置 Watermarkfrom pyspark.sql import functions as F # Define watermark on event time column df_with_watermark df \ .withWatermark(event_time, 10 minutes) # Watermark threshold: max_event_time - 10 minutes # Events older than watermark are dropped # State older than watermark is cleaned upwithWatermark只能作用于事件时间列TimestampType且必须在聚合之前设置。Watermark 时长选择指南ScenarioWatermark DurationReasoning实时分析1-5 分钟低延迟容忍极少量迟到数据标准 ETL10-30 分钟在延迟与迟到数据之间平衡迟到数据常见1-24 小时容纳明显延迟的事件尽力实时best-effort0 分钟不容忍迟到数据带 Watermark 的窗口聚合示例from pyspark.sql import functions as F from pyspark.sql.window import Window # Streaming aggregation with watermark result df \ .withWatermark(event_time, 10 minutes) \ .groupBy( F.window(event_time, 5 minutes, 1 minute), # 5-min tumbling window, 1-min slide user_id ) \ .agg( F.count(*).alias(event_count), F.sum(amount).alias(total_amount) ) # Output schema includes window struct: window.start, window.end query result \ .select( F.col(window.start).alias(window_start), F.col(window.end).alias(window_end), user_id, event_count, total_amount ) \ .writeStream \ .outputMode(append) \ .format(parquet) \ .option(path, s3://bucket/windowed_output/) \ .option(checkpointLocation, s3://bucket/checkpoints/) \ .start()窗口操作Tumbling / Sliding / Session窗口是流式聚合的基本时间容器。F.window默认按事件时间切分需与withWatermark配合。滚动窗口 Tumbling不重叠from pyspark.sql import functions as F # 5-minute tumbling windows result df \ .withWatermark(event_time, 10 minutes) \ .groupBy( F.window(event_time, 5 minutes), category ) \ .agg(F.sum(amount).alias(total)) # Windows: [00:00-00:05), [00:05-00:10), [00:10-00:15), ...滑动窗口 Sliding重叠滑动窗口window(time, windowDuration, slideDuration)每个 slide 步长生成一个与窗口长度相同的窗口适用于“最近 N 分钟”类滚动统计# 10-minute windows, sliding every 2 minutes result df \ .withWatermark(event_time, 10 minutes) \ .groupBy( F.window(event_time, 10 minutes, 2 minutes), category ) \ .agg(F.sum(amount).alias(total)) # Windows: [00:00-00:10), [00:02-00:12), [00:04-00:14), ...会话窗口 Session基于间隔会话窗口按“相邻事件间隔是否超过阈值”动态闭合窗口天然适配用户活跃会话、操作序列等场景F.session_window自 Spark 3.2 起可用# Session windows with 5-minute gap threshold result df \ .withWatermark(event_time, 10 minutes) \ .groupBy( F.session_window(event_time, 5 minutes), # Spark 3.2 user_id ) \ .agg( F.count(*).alias(events_in_session), F.first(event_time).alias(session_start), F.last(event_time).alias(session_end) )纵深说明窗口聚合依赖有界状态。参考文档给出的核心约束是——没有 Watermark 的聚合必然导致状态无限增长这一结论与 性能调优参考 中“启用 AQE、合理设置 shuffle 分区、监测 Spark UI 的 shuffle/spill/GC 指标”等规范共同构成流式作业的稳定性底线。有状态操作聚合、去重与自定义状态内置状态聚合按 key 的流式聚合groupBy(...).agg(...)由引擎维护状态状态随 Watermark 清理# Running count by key running_counts df \ .withWatermark(event_time, 1 hour) \ .groupBy(user_id) \ .agg(F.count(*).alias(total_events)) # State stored per user_id # Cleaned up based on watermark去重dropDuplicates在 Watermark 窗口内保留每个 key 的首次出现是最常见的幂等化手段# Drop duplicates within watermark window deduped df \ .withWatermark(event_time, 10 minutes) \ .dropDuplicates([event_id]) # Keep first occurrence # Can also dedupe by multiple columns deduped df \ .withWatermark(event_time, 10 minutes) \ .dropDuplicates([user_id, event_type, event_time])自定义有状态处理当内置聚合无法表达业务状态机时使用flatMapGroupsWithStateScala或其 PySpark 版本applyInPandasWithStateSpark 3.4。PySpark 版本通过GroupState读写每个 key 的自定义状态并支持显式设置超时按处理时间或事件时间# PySpark - Custom state using applyInPandasWithState (Spark 3.4) from pyspark.sql.streaming.state import GroupState, GroupStateTimeout def update_session_state( key: tuple, pdf_iter: Iterator[pd.DataFrame], state: GroupState ) - Iterator[pd.DataFrame]: # Get or initialize state if state.exists: session_data state.get else: session_data {count: 0, total: 0.0} # Process input data for pdf in pdf_iter: session_data[count] len(pdf) session_data[total] pdf[amount].sum() # Update state state.update(session_data) # Optionally set timeout state.setTimeoutDuration(10 * 60 * 1000) # 10 minutes # Yield output yield pd.DataFrame([{ user_id: key[0], event_count: session_data[count], total_amount: session_data[total] }]) # Apply stateful function result df \ .withWatermark(event_time, 10 minutes) \ .groupBy(user_id) \ .applyInPandasWithState( update_session_state, outputStructTypeoutput_schema, stateStructTypestate_schema, outputModeupdate, timeoutConfGroupStateTimeout.ProcessingTimeTimeout )Scala 端对应的flatMapGroupsWithState// Scala flatMapGroupsWithState import org.apache.spark.sql.streaming.{GroupState, GroupStateTimeout} case class UserState(count: Long, totalAmount: Double) case class UserOutput(userId: String, count: Long, totalAmount: Double) def updateState( userId: String, events: Iterator[Event], state: GroupState[UserState] ): Iterator[UserOutput] { val currentState state.getOption.getOrElse(UserState(0, 0.0)) var newCount currentState.count var newTotal currentState.totalAmount events.foreach { event newCount 1 newTotal event.amount } val newState UserState(newCount, newTotal) state.update(newState) state.setTimeoutDuration(10 minutes) Iterator(UserOutput(userId, newCount, newTotal)) } val result df .withWatermark(event_time, 10 minutes) .as[Event] .groupByKey(_.userId) .flatMapGroupsWithState( OutputMode.Update, GroupStateTimeout.ProcessingTimeTimeout )(updateState)实战提示自定义状态逻辑务必同时设置 Watermark 与超时setTimeoutDuration否则无法触发状态清理这两个参数共同决定了状态存储的上限。状态规模监控方法见下文“管理状态大小”。流式 JoinStream-Static Join流与静态表流式 DataFrame 与静态维度表如 Parquet 查找表Join无需 Watermark是最简单高效的流式关联方式# Join streaming data with static lookup table static_df spark.read.parquet(s3://bucket/lookup/) # Streaming df joined with static - no watermark needed result streaming_df.join(static_df, join_key, left) # Static table can be periodically refreshed # Use broadcast for small static tables from pyspark.sql.functions import broadcast result streaming_df.join(broadcast(static_df), join_key)小静态表用broadcast提示可避免 shuffle该写法与 spark-sql-dataframes.md 中“小于 200MB 的维度表优先广播”的 Join 策略一致。Stream-Stream Join流与流两路流 Join 必须两侧都设置 Watermark并且通常附加事件时间约束条件以限制需要缓存的“另一侧”数据量# Join two streams - requires watermarks on both from pyspark.sql import functions as F stream1 spark.readStream.format(kafka)... stream2 spark.readStream.format(kafka)... # Both streams need watermarks stream1_wm stream1.withWatermark(event_time, 10 minutes) stream2_wm stream2.withWatermark(event_time, 10 minutes) # Inner join with time constraint result stream1_wm.join( stream2_wm, F.expr( stream1.user_id stream2.user_id AND stream1.event_time stream2.event_time AND stream1.event_time stream2.event_time INTERVAL 5 MINUTES ), inner ) # Left outer join (Spark 2.3) result stream1_wm.join( stream2_wm, F.expr( stream1.user_id stream2.user_id AND stream1.event_time stream2.event_time - INTERVAL 5 MINUTES AND stream1.event_time stream2.event_time INTERVAL 5 MINUTES ), leftOuter )Join 类型支持矩阵Join TypeStream-StaticStream-StreamInnerYesYesLeft OuterYesYes (Spark 2.3)Right OuterYesYes (Spark 2.3)Full OuterYesYes (Spark 2.4)Left SemiYesNot supportedLeft AntiYesNot supported设计建议能拆成 Stream-Static 就不要做 Stream-Stream——后者要同时承担两路的 Watermark 与状态成本复杂度显著更高。输出 SinkKafka、文件、Delta Lake 与自定义Kafka Sink把处理结果写回 Kafka 时通常将某列作为 key、用F.to_json(F.struct(*))序列化整个行作为 value# Write to Kafka query df \ .select( F.col(user_id).alias(key), F.to_json(F.struct(*)).alias(value) ) \ .writeStream \ .format(kafka) \ .option(kafka.bootstrap.servers, broker1:9092) \ .option(topic, output_topic) \ .option(checkpointLocation, s3://bucket/checkpoints/) \ .start()File SinkParquet/JSON/CSV文件 Sink 支持partitionBy物理分区与trigger控制写盘节奏注意文件 Sink 通常只支持 append 输出模式# Parquet sink with partitioning query df.writeStream \ .format(parquet) \ .option(path, s3://bucket/output/) \ .option(checkpointLocation, s3://bucket/checkpoints/) \ .partitionBy(date, hour) \ .trigger(processingTime1 minute) \ .start() # JSON sink query df.writeStream \ .format(json) \ .option(path, s3://bucket/output/) \ .option(checkpointLocation, s3://bucket/checkpoints/) \ .start()Delta Lake SinkDelta Lake 提供 ACID 事务与 schema 演进mergeSchematrue且通过foreachBatch可构造真正的CDC Upsert# Delta Lake (ACID transactions, schema evolution) query df.writeStream \ .format(delta) \ .outputMode(append) \ .option(path, s3://bucket/delta_table/) \ .option(checkpointLocation, s3://bucket/checkpoints/) \ .option(mergeSchema, true) \ .start() # Upsert with foreachBatch def upsert_to_delta(batch_df, batch_id): delta_table DeltaTable.forPath(spark, s3://bucket/delta_table/) delta_table.alias(target).merge( batch_df.alias(source), target.id source.id ).whenMatchedUpdateAll() \ .whenNotMatchedInsertAll() \ .execute() query df.writeStream \ .foreachBatch(upsert_to_delta) \ .option(checkpointLocation, s3://bucket/checkpoints/) \ .start()自定义 SinkforeachBatchforeachBatch每批提供一个batch_df可用完整的批式 APIJDBC、自定义写入器处理并天然支持事务化写入def write_to_database(batch_df, batch_id): Write each micro-batch to external database. batch_df.write \ .format(jdbc) \ .option(url, jdbc:postgresql://host:5432/db) \ .option(dbtable, output_table) \ .option(user, user) \ .option(password, password) \ .mode(append) \ .save() query df.writeStream \ .foreachBatch(write_to_database) \ .option(checkpointLocation, s3://bucket/checkpoints/) \ .trigger(processingTime30 seconds) \ .start()foreach逐行回调foreach按行调用ForeachWriter的三阶段生命周期open → process → close适用于单行级别自定义处理但吞吐受限高吞吐场景优先选 foreachBatch# For custom processing of each row class ForeachWriter: def open(self, partition_id, epoch_id): # Initialize connection self.connection create_connection() return True def process(self, row): # Process each row self.connection.insert(row.asDict()) def close(self, error): # Clean up self.connection.close() query df.writeStream \ .foreach(ForeachWriter()) \ .start()Trigger控制批处理节奏Trigger 决定“多久触发一次微批次”直接权衡延迟与吞吐。可用 Trigger 类型# Process as fast as possible (default) query df.writeStream.trigger(processingTime0 seconds).start() # Fixed interval query df.writeStream.trigger(processingTime1 minute).start() # Once - process all available data, then stop query df.writeStream.trigger(onceTrue).start() # Available now - process all available data (Spark 3.3) query df.writeStream.trigger(availableNowTrue).start() # Continuous processing (experimental, low latency) query df.writeStream.trigger(continuous1 second).start()选择指南TriggerUse CaseprocessingTime0 seconds最大吞吐能多快就多快processingTimeN seconds受控的资源占用onceTrue类批处理的一次性执行availableNowTrue追赶积压数据Spark 3.3continuousN ms超低延迟实验特性监控与管理Query 生命周期管理writeStream.start()返回StreamingQuery句柄可用于读取元数据、等待与终止# Start query and get handle query df.writeStream.format(console).start() # Query properties print(fQuery ID: {query.id}) print(fRun ID: {query.runId}) print(fName: {query.name}) print(fIs Active: {query.isActive}) print(fStatus: {query.status}) print(fLast Progress: {query.lastProgress}) print(fRecent Progress: {query.recentProgress}) # Wait for termination query.awaitTermination() query.awaitTermination(timeout60) # With timeout # Stop query query.stop() # Get exception if failed exception query.exception()进度指标监控lastProgress携带吞吐、批次时长、状态大小等关键指标也可通过StreamingQueryListener订阅进度与终止事件# Get latest progress progress query.lastProgress if progress: print(fInput rows/sec: {progress[inputRowsPerSecond]}) print(fProcessed rows/sec: {progress[processedRowsPerSecond]}) print(fBatch ID: {progress[batchId]}) print(fDuration: {progress[batchDuration]} ms) print(fState rows: {progress[stateOperators]}) # Custom progress listener class ProgressListener: def onQueryProgress(self, event): print(fProgress: {event.progress}) def onQueryTerminated(self, event): print(fTerminated: {event.exception}) spark.streams.addListener(ProgressListener())Checkpoint容错的基石Checkpoint 目录必须设置它是故障恢复与精确一次语义的基础# Checkpoint location is required for fault tolerance query df.writeStream \ .format(parquet) \ .option(path, s3://bucket/output/) \ .option(checkpointLocation, s3://bucket/checkpoints/query_name/) \ .start()Checkpoint 中保存三类关键信息Offsets已处理到的数据偏移量进度State有状态操作的中间状态Commits已完成的批次记录。恢复机制查询重启后自动从最近 Checkpoint 续跑删除 Checkpoint 目录等于“干净启动”会丢失全部状态需谨慎操作。与 SKILL.md 的约束呼应其“MUST DO”清单要求监控 Spark UI 的 shuffle、spill、GC 指标并用生产规模数据验证性能目标而“MUST NOT DO”清单警告不要忽略 shuffle 分区调优、不要忽略 Spark UI 中的数据倾斜告警。这两组约束直接适用于流式作业的 Checkpoint 与状态监控环节。性能模式吞吐优化五板斧# 1. Increase Kafka partitions for parallelism # Consumer parallelism Kafka partitions # 2. Tune maxOffsetsPerTrigger单批数据量上限越大单批吞吐越高、批延迟越高 df spark.readStream \ .format(kafka) \ .option(maxOffsetsPerTrigger, 500000) \ .load() # 3. Optimize shuffle partitions spark.conf.set(spark.sql.shuffle.partitions, 100) # 4. Use appropriate trigger interval query df.writeStream \ .trigger(processingTime30 seconds) \ .start() # 5. Enable AQE for dynamic optimization spark.conf.set(spark.sql.adaptive.enabled, true)各点背后的机制如下Kafka 分区数决定消费并行度每个分区对应一个消费任务分区数不足是吞吐瓶颈的常见原因maxOffsetsPerTrigger是流式背压旋钮调大提升单批处理量调小控制峰值资源shuffle.partitions控制聚合/Join 的 shuffle 分区数其取值应与数据规模匹配默认 200 常常并非最优详见 partitioning-caching.md 的“24 分区/核心”经验法则AQESpark 3.x会自动合并小分区、处理倾斜 Join降低手工调参成本完整配置项见 performance-tuning.md 中的生产配置模板如spark.sql.adaptive.coalescePartitions.enabled、spark.sql.adaptive.skewJoin.enabled。管理状态大小有状态作业的稳定性 状态可控# 1. Always use watermarks for stateful operations df.withWatermark(event_time, 1 hour) # 2. Monitor state size in progress progress query.lastProgress for operator in progress[stateOperators]: print(fState rows: {operator[numRowsTotal]}) print(fMemory used: {operator[memoryUsedBytes]}) # 3. Configure state storeRocksDB 更擅长承载大状态 spark.conf.set(spark.sql.streaming.stateStore.providerClass, org.apache.spark.sql.execution.streaming.state.RocksDBStateStoreProvider) # 4. Set state cleanup mode spark.conf.set(spark.sql.streaming.stateStore.stateSchemaCheck, false)要点Watermark 是状态上限的“刹车”stateOperators指标行数与内存占用应纳入常规监控状态超大时切换 RocksDB 后端以缓解内存压力。对照 performance-tuning.md 的“内存压力症状表”——GC 时间过长、频繁 spill 到磁盘都是状态/缓存失控的信号处理方向是减少缓存、增大分区或提升内存。常见反模式对照# BAD: No watermark with aggregation df.groupBy(user_id).count() # Unbounded state growth! # GOOD: Always use watermark df.withWatermark(event_time, 1 hour).groupBy(user_id).count() # BAD: Complete mode with large state df.groupBy(user_id).count().writeStream.outputMode(complete) # Outputs entire state # GOOD: Update mode for incremental df.groupBy(user_id).count().writeStream.outputMode(update) # BAD: No checkpoint location query df.writeStream.format(console).start() # No fault tolerance! # GOOD: Always specify checkpoint query df.writeStream.format(console) \ .option(checkpointLocation, /checkpoints/query) \ .start() # BAD: foreach for high-throughput df.writeStream.foreach(process_row).start() # Row-by-row overhead # GOOD: foreachBatch for batched processing df.writeStream.foreachBatch(process_batch).start() # Batch-level efficiency四条反模式可总结为一句话有状态必加 Watermark、聚合慎用 Complete、永远配置 Checkpoint、自定义 Sink 优先 foreachBatch。最佳实践清单始终使用 Watermark—— 防止有状态聚合的状态无限增长选择合适的输出模式—— ETL 用 Append、聚合用 Update设置 Checkpoint 目录—— 容错恢复的前提自定义 Sink 用 foreachBatch 而非 foreach—— 批级处理的性能与事务能力远优于逐行监控状态大小—— 关注进度指标中的状态行数与内存调优触发间隔—— 在延迟与吞吐之间取平衡让 Kafka 分区数与并行度匹配—— 消费任务数 Kafka 分区数尽量用 Stream-Static Join—— 比 Stream-Stream 简单得多用生产数据速率测试—— 性能随数据量显著变化务必用接近生产的流量验证开启 Structured Streaming UI—— 在 Spark UI 中获取批粒度明细指标。参考与延伸阅读技能总览与使用约束skills/spark-engineer/SKILL.md本文主题源文档skills/spark-engineer/references/streaming-patterns.md批式与流式共用的调优参考skills/spark-engineer/references/performance-tuning.mdAQE、shuffle、状态/内存诊断决策树分区与缓存参考skills/spark-engineer/references/partitioning-caching.md分区数与分区大小经验法则DataFrame/SQL 与广播 Join 参考skills/spark-engineer/references/spark-sql-dataframes.mdRDD 底层操作参考skills/spark-engineer/references/rdd-operations.md流式作业的本质是在“延迟、吞吐、状态成本、容错”四者间做工程取舍用 Watermark 框住乱序与状态、用输出模式表达结果语义、用 Checkpoint 支撑精确一次、用 Trigger 和分区调优匹配流量节奏。把这套模式作为骨架再以生产规模数据反复验证即可构建稳定可靠的 Structured Streaming 生产管线。【免费下载链接】claude-skills67 Specialized Skills for Full-Stack Developers. Transform Claude Code into your expert pair programmer.项目地址: https://gitcode.com/GitHub_Trending/claud/claude-skills创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考