claude-skills 仓库的 Spark Engineer 技能全解PySpark 生产级开发、性能调优与 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本篇指南基于开源仓库 claude-skills67 个全栈开发者专项技能之一中的spark-engineer技能文档展开完整梳理其 5 阶段核心工作流、5 大参考主题DataFrame/SQL、RDD、分区与缓存、性能调优、流式计算并结合仓库中的技能结构与校验约定为读者呈现一份可直接复用的 Apache Spark / PySpark 生产级开发、优化与流式分析实战手册。读完本文你将掌握如何搭建可运行的 PySpark 数据管道、如何用 DataFrame API 写出高性能转换、如何诊断并解决数据倾斜与 shuffle 性能瓶颈以及如何基于 Structured Streaming 构建实时分析链路。技能定位何时调用 Spark Engineer在仓库的 SKILLS_GUIDE.md 中spark-engineer被归入「Data Machine Learning」分类决策路径为「Big Data Processing → Spark Engineer」并与 pandas-pro、ml-pipeline 组成数据与 ML 管道技能组合。从 skills/spark-engineer/SKILL.md 的 YAML frontmatter 可以看到该技能的完整元信息name:spark-engineerversion:1.1.0domain:data-mlscope:implementationoutput-format:codetriggers: Apache Spark、PySpark、Spark SQL、分布式计算、大数据、DataFrame API、RDD、Spark Streaming、structured streaming、数据分区、Spark 性能、集群计算、数据处理管道related-skills:python-pro、sql-pro、devops-engineer其description明确了调用时机编写 Spark 作业、调试性能问题、配置集群参数、处理.parquet文件、实现 RDD 管道、调优 shuffle、配置 executor 内存或构建 structured streaming 分析时都应启用该技能。它扮演的角色是资深 Apache Spark 工程师专注于高性能分布式数据处理、大规模 ETL 管道优化与生产级 Spark 应用构建。核心工作流从需求分析到验证的 5 个阶段原文档将 Spark 作业交付抽象为一个 5 阶段闭环这也是整个技能的行动骨架Analyze requirements分析需求—— 理解数据规模、转换逻辑、延迟要求与集群资源Design pipeline设计管道—— 选择 DataFrame 还是 RDD规划分区策略识别可广播的机会Implement实现—— 编写经过优化的 Spark 代码优化转换、适当缓存、正确的错误处理Optimize优化—— 分析 Spark UI调优 shuffle 分区数消除数据倾斜优化 join 与聚合Validate验证—— 继续操作前先在 Spark UI 中检查 shuffle spill用df.rdd.getNumPartitions()验证分区数量若检测到 spill 或倾斜返回第 4 步最后用生产规模数据测试监控资源使用验证性能目标。这是一个「优化-验证-回退」的迭代循环其验证标准无 spill、分区数合理、生产级数据规模测试会在下文各参考主题中反复出现。参考文档体系5 大主题的按需加载技能文档通过路由表将深入内容按场景延迟加载全部位于 skills/spark-engineer/references/ 目录主题参考文件加载时机Spark SQL 与 DataFramespark-sql-dataframes.mdDataFrame API、Spark SQL、schema、join、聚合RDD 操作rdd-operations.md转换、行动、pair RDD、自定义分区器分区与缓存partitioning-caching.md数据分区、持久化级别、广播变量性能调优performance-tuning.md配置、内存调优、shuffle 优化、倾斜处理流式模式streaming-patterns.mdStructured Streaming、watermark、有状态操作、sink这种「SKILL.md 主文档 references 深层文档」的结构是仓库全部技能的通用约定仓库的 scripts/validate-skills.py 会强制校验每个技能必须存在references/目录且至少含一个.md参考文件ReferencesDirectoryChecker、ReferenceFileCountChecker并校验正文引用的相对路径可解析ReferencePathChecker。换言之路由表中的每个文件名都经过 CI 校验可放心按表加载。实战代码四段可运行的核心示例1. Quick-Start 迷你管道PySpark原文档给出的一段可直接运行的入门管道涵盖会话构建、显式 schema、过滤聚合与写出前的分区验证from pyspark.sql import SparkSession from pyspark.sql import functions as F from pyspark.sql.types import StructType, StructField, StringType, LongType, DoubleType spark SparkSession.builder \ .appName(example-pipeline) \ .config(spark.sql.shuffle.partitions, 400) \ .config(spark.sql.adaptive.enabled, true) \ .getOrCreate() # 生产环境永远定义显式 schema schema StructType([ StructField(user_id, StringType(), False), StructField(event_ts, LongType(), False), StructField(amount, DoubleType(), True), ]) df spark.read.schema(schema).parquet(s3://bucket/events/) result df \ .filter(F.col(amount).isNotNull()) \ .groupBy(user_id) \ .agg(F.sum(amount).alias(total_amount), F.count(*).alias(event_count)) # 写出前验证分区数量 print(fPartition count: {result.rdd.getNumPartitions()}) result.write.mode(overwrite).parquet(s3://bucket/output/)注意两个开箱即用的配置spark.sql.shuffle.partitions400替代默认 200 的粗调与spark.sql.adaptive.enabledtrue开启 Spark 3.x 的 AQE 自适应优化这正是「Validate 阶段用 getNumPartitions 验证分区」的工作流落地。2. Broadcast Join小维表 200MBfrom pyspark.sql.functions import broadcast # Spark 会自动广播 dim_table加 hint 使意图显式化 enriched large_fact_df.join(broadcast(dim_df), onproduct_id, howleft)Broadcast 将整个小表复制到每个 executor 的内存中从而完全避免 shuffle。参考文档 spark-sql-dataframes.md 补充了其阈值语义默认自动广播阈值为 10MBspark.sql.autoBroadcastJoinThreshold可上调至 200MB设-1则强制关闭自动广播。可在 Spark UI 的 SQL Tab 中通过节点类型验证优化是否生效BroadcastHashJoin代表广播生效无 shuffleSortMergeJoin则说明发生了 shuffle。3. 用 Salting 处理数据倾斜当某个 join key 数据量远大于其他 key 时个别 task 会拖垮整个作业。Salting 的思路是给倾斜 key 加随机盐、同时把小表按盐值膨胀使数据均匀分布import pyspark.sql.functions as F SALT_BUCKETS 50 # 给大表加盐 skewed_df skewed_df.withColumn(salt, (F.rand() * SALT_BUCKETS).cast(int)) \ .withColumn(salted_key, F.concat(F.col(skewed_key), F.lit(_), F.col(salt))) # 小表按盐值膨胀explode other_df other_df.withColumn(salt, F.explode(F.array([F.lit(i) for i in range(SALT_BUCKETS)]))) \ .withColumn(salted_key, F.concat(F.col(skewed_key), F.lit(_), F.col(salt))) result skewed_df.join(other_df, onsalted_key, howinner) \ .drop(salt, salted_key)4. 正确的缓存模式# 仅当 DataFrame 被多次复用时才缓存 df_cleaned df.filter(...).withColumn(...).cache() df_cleaned.count() # 立即物化检查 Spark UI 是否有 spill report_a df_cleaned.groupBy(region).agg(...) report_b df_cleaned.groupBy(product).agg(...) df_cleaned.unpersist() # 用完释放缓存的关键纪律是缓存前先 count() 触发物化并在复用结束后主动unpersist()避免「缓存所有表而不衡量收益」的常见反模式。DataFrame 与 RDD如何选择参考文档 spark-sql-dataframes.md 与 rdd-operations.md 给出了清晰的决策标准使用 DataFrame处理结构化/半结构化数据JSON、Parquet、CSV、Avro执行类 SQL 操作join、聚合、过滤需要 Catalyst 优化器收益谓词下推、列裁剪使用列式存储以获得更好压缩。使用 RDD需要细粒度控制物理数据分布处理非结构化数据文本、自定义二进制格式实现自定义分区逻辑维护旧版 Spark 代码构建无法用 DataFrame 表达的自定义数据结构。核心原则是能用 DataFrame 就不用 RDD——Catalyst 与 Tungsten全阶段代码生成 WholeStageCodegen带来的优化是 RDD 无法企及的。Schema 定义生产环境必须显式显式 schema 是原文档反复强调的第一纪律。其收益有二避免spark.read.json()的自动推断需全量扫描数据同时保证类型安全。参考文档给出 PySpark 与 Scala 双版本示例from pyspark.sql.types import ( StructType, StructField, StringType, IntegerType, DoubleType, TimestampType, ArrayType, MapType ) user_schema StructType([ StructField(user_id, StringType(), nullableFalse), StructField(name, StringType(), nullableTrue), StructField(age, IntegerType(), nullableTrue), StructField(email, StringType(), nullableTrue), StructField(created_at, TimestampType(), nullableFalse), StructField(tags, ArrayType(StringType()), nullableTrue), StructField(metadata, MapType(StringType(), StringType()), nullableTrue) ]) df spark.read.schema(user_schema).json(s3://bucket/users/)若必须推断至少用samplingRatio抽样df spark.read.option(samplingRatio, 0.01).json(s3://bucket/users/)列操作、窗口函数与 Spark SQL参考文档系统整理了三大常用能力内置函数优先于 UDF字符串F.upper、F.split、日期时间F.year、F.date_format、F.datediff、数组F.size、F.array_contains、空值处理F.coalesce、F.isNotNull。原因很硬核Python UDF 相比内置函数慢10-100 倍且会在执行器上创建大量 Python 对象。窗口函数Window.partitionBy(...).orderBy(...)配合row_number/rank/dense_rank、lag/lead/sum以及rangeBetween实现 7 天滚动窗口。Spark SQLcreateOrReplaceTempView会话级/createOrReplaceGlobalTempView应用级需以global_temp.前缀访问支持 CTE、子查询与PERCENT_RANK等分析函数。Catalyst 优化器三剑客参考文档总结了利用 Catalyst 的三种裁剪谓词下推spark.read.parquet(...).filter(...)会把过滤下推到数据源PushedFilters用df.explain(True)验证物理计划列裁剪select(id, name, amount)只读所需列先全量读再 select 是反模式分区裁剪数据按日期分区后filter(date between ...)只读匹配分区Spark UI 中 Files Read 应显著减少。聚合与透视groupBy().agg()支持count/sum/avg/min/max/stddev/countDistinct等注意collect_list可能 OOM慎用。多组集合可用GROUPING SETS、rollup与cubepivot可将行值转列配合expr(stack(...))可实现 unpivot。Join 策略类型、广播与倾斜六种 join 类型参考文档给出完整对照inner仅匹配记录、left/right外连接、full全外连接、left_anti左表不在右表、left_semi存在匹配但不取右表列、crossJoin笛卡尔积慎用。倾斜 join 的三种解法参考文档 spark-sql-dataframes.md 提供了 AQE 自动处理与手动加盐两条路径# 方式一AQE 自动处理Spark 3.0 spark.conf.set(spark.sql.adaptive.enabled, true) spark.conf.set(spark.sql.adaptive.skewJoin.enabled, true) spark.conf.set(spark.sql.adaptive.skewJoin.skewedPartitionFactor, 5) spark.conf.set(spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes, 256MB)# 方式二手动加盐大表取模 小表 explode salt_count 10 large_df_salted large_df.withColumn( join_key_salted, F.concat(F.col(join_key), F.lit(_), (F.monotonically_increasing_id() % salt_count).cast(string)) ) small_df_exploded small_df.withColumn( salt, F.explode(F.array([F.lit(i) for i in range(salt_count)])) ).withColumn( join_key_salted, F.concat(F.col(join_key), F.lit(_), F.col(salt).cast(string)) ) result large_df_salted.join(small_df_exploded, join_key_salted)RDD 操作深度转换、行动与自定义分区器惰性转换Lazy Transformationsmap、flatMap、filter、distinct触发 shuffle、sample、union/intersection/subtract后两者触发 shuffle、cartesian极昂贵。批量处理优先mapPartitions——数据库连接等昂贵初始化每分区只做一次def process_partition(iterator): connection create_database_connection() # 每分区初始化一次 try: for record in iterator: yield connection.process(record) finally: connection.close() result_rdd rdd.mapPartitions(process_partition)Pair RDDreduceByKey 优于 groupByKey这是 RDD 性能的第一原则groupByKey()会把所有 value 全量 shuffle 到对应分区内存密集、可能 OOM而reduceByKey先在本地合并再 shuffle数据量小得多。同样地aggregateByKey/combineByKey通过分区内/跨分区合并函数实现单次 shuffle 的多级聚合。Actions 的 OOM 纪律collect()将全部数据拉回 driver是大数据集 OOM 的头号原因。替代方案take(n)取前 n 条、takeSample随机抽样、foreachPartition分布式处理、saveAsTextFile分布式写出。自定义分区器partitionBy默认使用 HashPartitioner自定义分区器可继承pyspark.Partitioner或 Scala 的org.apache.spark.Partitioner实现numPartitions()与getPartition(key)。关键语义mapValues/flatMapValues保留分区器map会丢失分区器——前者让连续转换无需重新 shuffle。广播变量与累加器广播变量将只读字典分发到所有 executorlookup_broadcast.value访问用毕unpersist()destroy()累加器longAccumulator统计错误数、自定义AccumulatorParam收集错误类型。警告task 重试时累加器可能被多次更新只能用于调试/监控不可用于业务逻辑。分区与缓存并行度与内存的平衡艺术分区数量与大小指南参考文档 partitioning-caching.md 给出的黄金规则每 CPU 核心 2-4 个分区100 个 executor 核心对应 200-400 个分区单分区目标大小 128MB-256MB100GB 数据约 800 个分区用df.rdd.getNumPartitions()检查当前分区数。数据量目标分区大小分区数 1GB64MB8-161-10GB128MB8-8010-100GB128-256MB40-800100GB-1TB256MB400-4000 1TB256MB4000repartition 与 coalesce 的分工repartition(n)/repartition(col)全量 shuffle用于增加分区、按列分布同 key 同分区、实现均匀分布coalesce(n)只减少分区且避免全量 shuffle典型场景是 filter 后数据大幅缩减如从 200 降到 40repartitionByRange(n, col)按范围分区适合有序访问。反模式警示repartition应在filter之后做先过滤再压缩先 repartition 再 filter 等于「先 shuffle 后裁剪」。可用F.spark_partition_id()统计各分区行数分布max/avg 3即视为倾斜。Shuffle 分区数与 AQEspark.sql.shuffle.partitions默认 200 常常不合适小数据10GB可降到 50大数据100GB可提到 2000。Spark 3.x 的 AQE 能动态调整spark.conf.set(spark.sql.adaptive.enabled, true) spark.conf.set(spark.sql.adaptive.coalescePartitions.enabled, true) spark.conf.set(spark.sql.adaptive.coalescePartitions.minPartitionSize, 64MB) spark.conf.set(spark.sql.adaptive.advisoryPartitionSizeInBytes, 128MB) spark.conf.set(spark.sql.adaptive.skewJoin.enabled, true) spark.conf.set(spark.sql.adaptive.localShuffleReader.enabled, true)持久化级别选择cache()等价于persist(StorageLevel.MEMORY_AND_DISK)。各存储级别适用场景存储级别适用场景MEMORY_ONLY内存充足需最快访问MEMORY_AND_DISK默认值大多数场景安全MEMORY_ONLY_SER内存受限、CPU 可用序列化省内存MEMORY_AND_DISK_SER大数据 内存受限DISK_ONLY超大且内存稀缺OFF_HEAP使用 Tungsten 堆外内存缓存纪律只在 DataFrame 被多次复用、计算昂贵复杂 join/聚合、迭代算法或 Notebook 交互探索时缓存缓存后立即count()物化用完unpersist()。长链路管道可用df.checkpoint()截断血统。Spark UI 的 Storage Tab 中Fraction Cached应达到 100%否则说明内存不足。文件分区与 BucketingHive 风格分区写df.write.partitionBy(year, month).parquet(...)生成year2024/month01/目录读取时分区列自动从路径发现过滤即分区裁剪Bucket 分桶bucketBy(100, user_id).sortBy(timestamp).saveAsTable(...)同桶数表 join 免 shuffle。注意Bucketing 需要 Hive metastore 与saveAsTable不支持直接文件写。性能调优从集群配置到问题定位Executor 配置与规模参考spark.conf.set(spark.executor.instances, 10) # executor 数量 spark.conf.set(spark.executor.cores, 4) # 每 executor 核心数 spark.conf.set(spark.executor.memory, 16g) # 每 executor 内存 # 动态分配推荐用于波动负载 spark.conf.set(spark.dynamicAllocation.enabled, true) spark.conf.set(spark.dynamicAllocation.minExecutors, 2) spark.conf.set(spark.dynamicAllocation.maxExecutors, 100)集群规模Executor 内存Executor 核心实例数小型开发4-8GB2-42-5中型8-16GB4-510-50大型16-32GB5-850-200超大型32-64GB8-16200经验法则每 executor 5 核心最优避免 HDFS I/O 瓶颈每节点预留 1 核心给 OS/YARN预留 1GB 节点开销executor.memoryOverhead max(384MB, executor.memory × 10%)。内存比例一般保持默认spark.memory.fraction0.6storageFraction0.5。Shuffle 优化减量 压缩spark.conf.set(spark.sql.shuffle.partitions, 200) # 按数据量调整 spark.conf.set(spark.shuffle.compress, true) spark.conf.set(spark.io.compression.codec, lz4) # 快速压缩 spark.conf.set(spark.shuffle.file.buffer, 64k) spark.conf.set(spark.shuffle.io.maxRetries, 3) spark.conf.set(spark.shuffle.io.retryWait, 5s)减少 shuffle 体积的 5 招join/聚合前先 filter小表用 broadcast完全免 shuffleshuffle 前只 select 必要列不要拖 50 列RDD 用 reduceByKey 而非 groupByKeyfilter 数据缩减后 coalesce 降分区。数据倾斜的识别与四套解法识别groupBy(join_key).count()查看 key 分布max/avg 10视为严重倾斜Spark UI 中少数 task 耗时时长远超中位数Straggler。AQE 自动处理开启spark.sql.adaptive.skewJoin.enabledSpark 自动拆分倾斜分区Salting加盐对倾斜 key 加随机盐、小表 explode均匀打散倾斜 key 单独广播将数据拆为倾斜部分用 broadcast join与正常部分普通 join再 union极端单 key 迭代广播单 key 数据用crossJoin(broadcast(...))处理其余走正常路径。内存调优与 GC内存压力症状速查长时间 GC 停顿 → 缓存过多减少缓存或改序列化存储spill 到磁盘 → 分区过大增加分区数或内存driver OOM → 大 collect/broadcastexecutor OOM → 分区过大repartition 或加内存。GC 目标GC Time task 总时间的 10%。可用 G1GC 参数调优-XX:UseG1GC -XX:InitiatingHeapOccupancyPercent35。Join 策略与 HintSpark 3.0Broadcast Hash Join小表200MB→broadcast()Sort Merge Join大表等值 join 的默认策略Shuffle Hash Join中型表 内存受限Bucket Join预分桶表免 shuffle。Spark 3.0 支持显式 hintdf2.hint(broadcast)、df1.hint(merge)、df1.hint(shuffle_hash)、df1.hint(shuffle_replicate_nl)。用explain(True)检查物理计划中是否出现BroadcastNestedLoopJoin昂贵应避免或CartesianProduct除非有意为之。I/O 优化与小文件问题读Parquet 为 Spark 最优格式spark.sql.parquet.filterPushdowntruemergeSchemafalseschema 一致时更快列裁剪 分区裁剪 显式 schema 三件套写目标文件 128-256MBdf.coalesce(n).write.parquet(...)做小文件合并partitionBy分区写bucketBy分桶写需 Hive metastoreoption(compression, snappy)指定压缩小文件问题大量小文件导致 task 调度开销与 NameNode 压力检测后用coalesce/repartition重写压缩。Spark UI 五 Tab 体检表Tab关键指标异常时动作JobsJob 时长、阶段数阶段越多 shuffle 越多Stages时长 5min/阶段、Shuffle Read Blocked Time≈0、Spill(Disk)0、GC10%拆分大阶段、排查网络/内存/倾斜ExecutorsStorage Memory、GC Time、失败任务调整内存/缓存SQLDuration、物理计划细节确认 BroadcastExchange/ShuffleExchangeStorageFraction Cached 应 100%增加内存或减少缓存生产配置模板可直接套用spark_configs { # Executor 配置 spark.executor.instances: 50, spark.executor.cores: 5, spark.executor.memory: 16g, spark.executor.memoryOverhead: 2g, # Driver 配置 spark.driver.memory: 8g, spark.driver.maxResultSize: 4g, # Shuffle 配置 spark.sql.shuffle.partitions: 500, spark.shuffle.compress: true, spark.io.compression.codec: lz4, # SQL 优化 spark.sql.adaptive.enabled: true, spark.sql.adaptive.coalescePartitions.enabled: true, spark.sql.adaptive.skewJoin.enabled: true, spark.sql.autoBroadcastJoinThreshold: str(200 * 1024 * 1024), # 200MB # 序列化 spark.serializer: org.apache.spark.serializer.KryoSerializer, # 动态分配 spark.dynamicAllocation.enabled: true, spark.dynamicAllocation.minExecutors: 5, spark.dynamicAllocation.maxExecutors: 100, } for key, value in spark_configs.items(): spark.conf.set(key, value)排障决策树Slow Spark Job ├── GC 时间过长 (10%)? → 增加 executor 内存或减少缓存 ├── Shuffle spill 到磁盘? → 增加分区数或内存 ├── Task 时长不均? → 数据倾斜用 salting 或 AQE ├── Shuffle 读取时间长? → 网络瓶颈提高数据本地性 ├── Shuffle 体积过大? → 提前过滤、广播小表 └── 小任务过多? → 用 coalesce 减少分区Structured Streaming实时分析实践适用边界参考文档 streaming-patterns.md 明确适用于连续数据流Kafka、文件、socket、需要 exactly-once 保证、实时分析/看板、事件驱动架构与流式增量 ETL若批处理已足够复杂度更低、需要亚秒级延迟考虑 Flink或仅简单事件处理Kafka Streams 可能足够则考虑替代方案。三种输入源KafkareadStream.format(kafka)配置kafka.bootstrap.servers、subscribe、startingOffsets、maxOffsetsPerTrigger与 SASL 安全参数value 为字节流需用F.from_json解析文件源readStream.format(parquet/json/csv)pathmaxFilesPerTrigger自动发现新文件Rate 源测试用format(rate)按rowsPerSecond生成测试数据。三种输出模式的选择使用场景输出模式说明ETL 到文件append默认高效窗口聚合append需配合 watermark运行计数/求和update增量输出需要完整状态的看板complete昂贵去重append配 dropDuplicatesWatermark有状态操作的生死线Watermark 定义迟到数据可被丢弃的阈值是 Spark 清理旧状态、按正确时间输出结果、处理乱序事件的关键df_with_watermark df.withWatermark(event_time, 10 minutes)Watermark 时长的经验参考实时分析 1-5 分钟标准 ETL 10-30 分钟迟到数据常见时 1-24 小时尽力而为的实时设为 0。最严重的流式反模式是无 watermark 的聚合——状态无界增长最终 OOM。groupBy(...).count()永远要配withWatermark。三类窗口与有状态操作Tumbling滚动F.window(event_time, 5 minutes)无重叠Sliding滑动F.window(event_time, 10 minutes, 2 minutes)有重叠Session会话F.session_window(event_time, 5 minutes)Spark 3.2按活动间隙分窗去重withWatermark dropDuplicates([event_id])自定义状态PySpark 用applyInPandasWithStateSpark 3.4Scala 用flatMapGroupsWithState支持GroupState状态读写与setTimeoutDuration超时。流式 Join 与 Sink流-静态 join无需 watermark静态表可周期性刷新小表加broadcast流-流 join两侧都必须设 watermark 并加时间约束event_time区间条件支持矩阵为Inner/Left Outer/Right Outer/Full Outer 全支持Left Semi/Left Anti 的流-流版本不支持Sink 家族Kafka sinkF.to_json(F.struct(*))编码 value、文件 sinkparquet/json/csv partitionBy、Delta sinkforeachBatch做 upsert merge、foreachBatch自定义 sink如 JDBC 批量写、foreach逐行低吞吐场景。优先 foreachBatch 而非 foreach——后者逐行开销巨大。Trigger 选择processingTime0 seconds最大吞吐默认、processingTimeN seconds受控资源、onceTrue批式处理一轮后停止、availableNowTrueSpark 3.3追赶处理、continuous1 second实验性超低延迟。监控与 Checkpointquery.lastProgress提供inputRowsPerSecond、processedRowsPerSecond、batchId、stateOperators状态行数与内存等指标spark.streams.addListener可注册自定义进度监听器。Checkpoint 是容错的必需品记录 offsets、state、commits每次writeStream都必须指定checkpointLocation注意删除 checkpoint 目录 丢失全部状态。大状态可切换 RocksDB 状态存储RocksDBStateStoreProvider。约束红线MUST DO 与 MUST NOT DO原文档以两条清单界定了技能的行为边界这也是判断代码质量的金标准MUST DO必须做结构化数据处理用 DataFrame API 而非 RDD生产管道定义显式 schema合理分区每 executor 核心 200-1000 个分区中间结果仅在多次复用时缓存小维表200MB用广播 join用 salting 或自定义分区处理数据倾斜监控 Spark UI 的 shuffle、spill 与 GC 指标用生产级数据量测试。MUST NOT DO禁止做对大数据集使用collect()导致 OOM生产环境跳过 schema 定义依赖推断不衡量收益就缓存所有 DataFrame忽略 shuffle 分区调优默认 200 常常是错的有内置函数时用 UDF慢 10-100 倍不 coalesce 就处理小文件小文件问题不理解惰性求值就编写转换忽略 Spark UI 中的数据倾斜告警。输出模板与知识覆盖当调用该技能实现 Spark 方案时交付物应包含 5 部分完整的 Spark 代码PySpark 或 Scala带类型标注配置建议executors、内存、shuffle 分区数分区策略说明性能分析预期 shuffle 体积、内存用量监控建议需重点观察的 Spark UI 指标。知识域覆盖Spark DataFrame API、Spark SQL、RDD 转换/行动、Catalyst 优化器、Tungsten 执行引擎、分区策略、广播变量、累加器、structured streaming、watermark、checkpointing、Spark UI 分析、内存管理、shuffle 优化。在仓库中的落地技能结构与质量保障该技能遵循仓库统一的技能规范可在 CLAUDE.md 与 SKILLS_GUIDE.md 中查看约定由 scripts/validate-skills.py 在 CI 中强制校验SKILL.md 必须以 YAML frontmatter 开头、包含name/description必填字段、metadata 必须含triggers/role/scope/output-format/domain/related-skills子字段、必须存在references/目录、Core Workflow 须含 5 个编号步骤。发布记录见 CHANGELOG.md技能版本管理见 version.json。小结Spark Engineer 技能文档围绕「分析需求 → 设计管道 → 实现 → 优化 → 验证」的五步闭环沉淀了从 DataFrame/RDD 选型、显式 schema、join 优化、倾斜处理、分区缓存到 Structured Streaming 的完整实战知识。其价值不在于某一条孤立技巧而在于把 Spark 生产中反复踩坑的经验固化为可执行的纪律显式 schema 不推断、内置函数不写 UDF、reduceByKey 不 groupByKey、流式聚合必有 watermark、小表必广播、缓存必物化必释放、性能必以 Spark UI 与生产级数据验证。将这套方法论与仓库中的 5 篇参考文档配合使用即可获得一位资深 Spark 工程师的完整决策框架。【免费下载链接】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），仅供参考