---

name: spark-optimization description: "使用分区、缓存、Shuffle 优化和内存调优来优化 Apache Spark 作业。在改善 Spark 性能、调试慢作业或扩展数据处理管道时使用。" risk: unknown source: community date_added: "2026-02-27"

Apache Spark 优化

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

不适用场景

• 任务与 Apache Spark 优化无关
• 需要此范围之外的不同领域或工具

说明

• 明确目标、约束和所需输入。
• 应用相关最佳实践并验证结果。
• 提供可操作的步骤和验证。
• 如需详细示例，打开 resources/implementation-playbook.md。

何时使用此技能

• 优化慢 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 | 最小化宽转换 | | 数据倾斜 | 任务时长不均 | 加盐、广播 Join | | 序列化 | 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/")

模式

模式 1：最优分区

# Calculate optimal partition count
def calculate_partitions(data_size_gb: float, partition_size_mb: int = 128) -> int:
    """
    Optimal partition size: 128MB - 256MB
    Too few: Under-utilization, memory pressure
    Too many: Task scheduling overhead
    """
    return max(int(data_size_gb * 1024 / partition_size_mb), 1)

# Repartition for even distribution
df_repartitioned = df.repartition(200, "partition_key")

# Coalesce to reduce partitions (no shuffle)
df_coalesced = df.coalesce(100)

# Partition pruning with predicate pushdown
df = (spark.read.parquet("s3://bucket/data/")
    .filter(F.col("date") == "2024-01-01"))  # Spark pushes this down

# Write with partitioning for future queries
(df.write
    .partitionBy("year", "month", "day")
    .mode("overwrite")
    .parquet("s3://bucket/output/"))