skillhub-110-spark-optimization/SKILL.md

name, description

name	description
spark-optimization	优化 Apache Spark 作业，涵盖分区、缓存、shuffle 优化与内存调优。适用于提升 Spark 性能、调试慢作业或扩缩数据处理管道时使用。

Apache Spark 优化

优化 Apache Spark 作业的生产级模式，包括分区策略、内存管理、shuffle 优化与性能调优。

何时使用此技能

• 优化运行缓慢的 Spark 作业
• 调优内存与执行器配置
• 实现高效的分区策略
• 调试 Spark 性能问题
• 为大数据集扩缩 Spark 管道
• 减少 shuffle 与数据倾斜

核心概念

1. Spark 执行模型

Driver 程序
    ↓
作业（由 action 触发）
    ↓
阶段（以 shuffle 分隔）
    ↓
任务（每个分区一个）

2. 关键性能因素

因素	影响	解决方案
Shuffle	网络 I/O、磁盘 I/O	减少宽依赖转换
数据倾斜	任务执行时间不均	加盐、广播连接
序列化	CPU 开销	使用 Kryo、列式格式
内存	GC 压力、溢出到磁盘	调优执行器内存
分区数	并行度	设置合适的分区大小

快速开始

from pyspark.sql import SparkSession
from pyspark.sql import functions as F

# 创建优化后的 Spark 会话
spark = (SparkSession.builder
    .appName("OptimizedJob")
    .config("spark.sql.adaptive.enabled", "true")
    .config("spark.sql.adaptive.coalescePartitions.enabled", "true")
    .config("spark.sql.adaptive.skewJoin.enabled", "true")
    .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
    .config("spark.sql.shuffle.partitions", "200")
    .getOrCreate())

# 使用优化设置读取数据
df = (spark.read
    .format("parquet")
    .option("mergeSchema", "false")
    .load("s3://bucket/data/"))

# 高效转换
result = (df
    .filter(F.col("date") >= "2024-01-01")
    .select("id", "amount", "category")
    .groupBy("category")
    .agg(F.sum("amount").alias("total")))

result.write.mode("overwrite").parquet("s3://bucket/output/")

详细模式与示例

详细模式文档位于 references/details.md。当上述导航层级不够用时，请阅读该文件。

最佳实践

应该做的

• 启用 AQE — 自适应查询执行能处理许多问题
• 使用 Parquet/Delta — 列式格式并启用压缩
• 广播小表 — 对小表连接避免 shuffle
• 监控 Spark UI — 检查数据倾斜、溢出、GC 情况
• 设置合适的分区大小 — 每个分区 128MB — 256MB

不应做的

• 不要收集大数据 — 保持数据分布式
• 不要不必要地使用 UDF — 优先使用内置函数
• 不要过度缓存 — 内存是有限的
• 不要忽视数据倾斜 — 它会主导作业运行时间
• 不要用 .count() 判断是否存在 — 应使用 .take(1) 或 .isEmpty()