mcpField
S
spark-optimization

spark-optimization

github.com/full-stack-skills数据与数据库147 次安装4 周前更新
代码仓库

关于 spark-optimization

这是一个提供 Apache Spark 作业调优与故障排查实践的 skill，面向需要处理性能、内存、分区、shuffle 和数据倾斜的从业者。

主要功能

  1. 01分区大小策略
  2. 02内存与 executor 调优
  3. 03降低 shuffle 和数据倾斜
  4. 04Spark 执行模型概览
  5. 05references/details.md 中的实践示例

适用场景

  1. 01诊断运行缓慢的 Spark 作业
  2. 02调整 executor 内存与配置
  3. 03扩展大型数据集处理流水线
  4. 04减少 shuffle 与数据倾斜

安装

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

$npx skills add https://github.com/full-stack-skills/devops-skills --skill spark-optimization

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

包含的文件

除 SKILL.md 外，安装时一起下载的参考资料和脚本。

LICENSE.txt
agents/
openai.yaml
examples/
advanced-usage.md
basic-usage.md
integration.md
references/
details.md

详细介绍

来源声明：本技能内容源自 wshobson/agents （MIT License，Copyright (c) 2024 Seth Hobson；完整许可文本见同目录 LICENSE.txt）。 使用、修改与再分发须遵守该 MIT 许可条款并保留版权声明。

Apache Spark 优化

优化 Apache Spark 作业的生产实践，包括分区策略、内存管理、Shuffle 优化和性能调优。

何时使用此技能

  • 优化运行缓慢的 Spark 作业
  • 调优内存和执行器配置
  • 实现高效的分区策略
  • 调试 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尽量减少宽转换
数据倾斜任务耗时不均加盐、广播连接
序列化CPU 开销使用 Kryo、列式格式
内存GC 压力、溢写调优执行器内存
分区并行度合理设置分区大小

快速入门

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 - 采用带压缩的列式格式
  • 广播小表 - 避免小型连接中的 Shuffle
  • 监控 Spark UI - 检查数据倾斜、溢写和 GC
  • 合理设置分区大小 - 每个分区 128MB - 256MB
不应该做
  • 不要收集大量数据 - 保持数据分布式处理
  • 不要不必要地使用 UDF - 使用内置函数
  • 不要过度缓存 - 内存有限
  • 不要忽略数据倾斜 - 它会占用作业的大部分时间
  • 不要用 .count() 判断是否存在 - 使用 .take(1) 或 .isEmpty()

评论

支持 Markdown