mcpField
S
spark-optimization

spark-optimization

github.com/wshobson数据与数据库41.9k10.8k 次安装2 周前更新
代码仓库

关于 spark-optimization

用于优化 Apache Spark 作业的技能，提供分区、内存、Shuffle 和性能排查方面的生产实践，适合调优或扩展 Spark 数据流水线的团队。

主要功能

  1. 01分区大小规划
  2. 02内存与 Executor 调优
  3. 03减少 Shuffle 的方法
  4. 04缓解数据倾斜
  5. 05Spark 性能问题排查

适用场景

  1. 01加速运行缓慢的 Spark 作业
  2. 02调整 Executor 内存设置
  3. 03扩展大型数据集流水线
  4. 04排查数据倾斜与 Shuffle 开销

安装

复制下面的命令，在终端中运行。

$npx skills add https://github.com/wshobson/agents --skill spark-optimization

安装后在 Claude Code 中输入/spark-optimization确认已加载。

附带文件references/details.md

详细介绍

Apache Spark 优化

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

何时使用本技能

  • 优化运行缓慢的 Spark 作业
  • 调优内存和 executor 配置
  • 实施高效的分区策略
  • 调试 Spark 性能问题
  • 为大型数据集扩展 Spark 流水线
  • 减少 shuffle 和数据倾斜

核心概念

1. Spark 执行模型
Driver Program
    ↓
Job (triggered by action)
    ↓
Stages (separated by shuffles)
    ↓
Tasks (one per partition)
2. 关键性能因素
因素影响解决方案
Shuffle网络 I/O、磁盘 I/O尽量减少宽转换
数据倾斜任务时长不均加盐、广播 join
序列化CPU 开销使用 Kryo、列式格式
内存GC 压力、溢出调优 executor 内存
分区并行度合理设置分区大小

快速开始

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

# Create optimized Spark session
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())

# Read with optimized settings
df = (spark.read
    .format("parquet")
    .option("mergeSchema", "false")
    .load("s3://bucket/data/"))

# Efficient transformations
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 - 带压缩的列式格式
  • 广播小表 - 小表 join 时避免 shuffle
  • 监控 Spark UI - 检查倾斜、溢出、GC
  • 合理设置分区大小 - 每个分区 128MB - 256MB
避免做法
  • 不要 collect 大数据 - 保持数据分布式处理
  • 不要不必要地使用 UDF - 使用内置函数
  • 不要过度缓存 - 内存有限
  • 不要忽视数据倾斜 - 它会主导作业耗时
  • 不要用 .count() 判断存在性 - 使用 .take(1) 或 .isEmpty()

评论

支持 Markdown