Spark SQL 性能调优大师

> 从数据倾斜到 Shuffle 优化，从 Spark UI 诊断到 AQE 自适应——"让每个 Spark Job 都跑出最优性能"。

一、Spark 性能调优全景图

                    ┌──────────────────────┐
                    │   Spark 性能调优      │
                    └──────────┬───────────┘
                               │
     ┌─────────────┬───────────┼───────────┬─────────────┐
     ▼             ▼           ▼           ▼             ▼
  数据倾斜      Shuffle     Join策略    内存管理      AQE自适应
  诊断修复      优化         选择         调优          执行

二、快速开始

# 启动 Spark 集群 + Jupyter 环境
docker-compose up -d

# 访问 Spark Master UI: http://localhost:8080
# 访问 Jupyter: http://localhost:8888 (token: spark)

# 运行性能诊断脚本
python scripts/diagnose_skew.py --table orders --key user_id

# 运行调优建议生成器
python scripts/tuning_advisor.py --app-id app-20260601-0001

三、数据倾斜：7种场景诊断与修复

3.1 倾斜诊断

# scripts/diagnose_skew.py
"""自动诊断数据倾斜：分析 key 分布，识别倾斜 key"""

def diagnose_skew(spark, table, key_column):
    """诊断指定表的指定 key 是否存在数据倾斜"""
    df = spark.table(table)

    # 统计 key 分布
    key_dist = df.groupBy(key_column).count()

    # 计算倾斜指标
    stats = key_dist.select(
        F.count(key_column).alias('total_keys'),
        F.max('count').alias('max_count'),
        F.min('count').alias('min_count'),
        F.avg('count').alias('avg_count'),
        F.stddev('count').alias('stddev_count'),
        F.percentile_approx('count', 0.99).alias('p99_count'),
    ).collect()[0]

    skew_ratio = stats['max_count'] / stats['avg_count']
    is_skewed = skew_ratio > 10  # 最大key数据量超过平均10倍

    # 找出倾斜 key
    threshold = stats['avg_count'] * 5
    skewed_keys = key_dist.filter(F.col('count') > threshold)

    return {
        'is_skewed': is_skewed,
        'skew_ratio': round(skew_ratio, 2),
        'skewed_key_count': skewed_keys.count(),
        'top_skewed_keys': skewed_keys.orderBy(F.desc('count')).limit(10).collect(),
        'recommendation': get_skew_fix_recommendation(skew_ratio)
    }

3.2 七种场景与修复方案

场景1：单 key 数据量极大（如 null 值倾斜）

# 方案：加盐（Salting）
def salt_join(df1, df2, join_key, salt_num=100):
    """对倾斜 key 加随机盐，分散到多个分区"""
    df1_salted = df1.withColumn(
        'salt', (F.rand() * salt_num).cast('int')
    ).withColumn(
        'salted_key', F.concat(F.col(join_key), F.lit('_'), F.col('salt'))
    )

    # df2 需要膨胀 salt_num 倍
    df2_exploded = df2.crossJoin(
        spark.range(0, salt_num).toDF('salt')
    ).withColumn(
        'salted_key', F.concat(F.col(join_key), F.lit('_'), F.col('salt'))
    )

    return df1_salted.join(df2_exploded, 'salted_key')

场景2：大表 join 小表（Broadcast Join）

# 方案：显式 Broadcast Hint
from pyspark.sql.functions import broadcast

result = large_df.join(
    broadcast(small_df),  # 小表广播到所有 Executor
    'join_key',
    'left'
)

# 自动 Broadcast 阈值
spark.conf.set('spark.sql.autoBroadcastJoinThreshold', 10 * 1024 * 1024)  # 10MB

场景3：两张大表 join 都倾斜

# 方案：分离倾斜 key 单独处理
def split_skew_join(df1, df2, join_key, skew_keys):
    """将倾斜 key 和非倾斜 key 分开 join"""
    # 倾斜部分：加盐 join
    df1_skew = df1.filter(F.col(join_key).isin(skew_keys))
    df2_skew = df2.filter(F.col(join_key).isin(skew_keys))
    result_skew = salt_join(df1_skew, df2_skew, join_key)

    # 非倾斜部分：正常 join
    df1_normal = df1.filter(~F.col(join_key).isin(skew_keys))
    df2_normal = df2.filter(~F.col(join_key).isin(skew_keys))
    result_normal = df1_normal.join(df2_normal, join_key)

    return result_skew.union(result_normal)

场景4：GroupBy 聚合倾斜

# 方案：两阶段聚合（局部聚合 + 全局聚合）
def two_phase_agg(df, group_key, agg_col):
    """先加盐局部聚合，再去盐全局聚合"""
    salt_num = 100

    # Phase 1: 加盐局部聚合
    phase1 = df.withColumn('salt', (F.rand() * salt_num).cast('int')) \
               .groupBy(group_key, 'salt') \
               .agg(F.sum(agg_col).alias('partial_sum'),
                    F.count(agg_col).alias('partial_count'))

    # Phase 2: 去盐全局聚合
    phase2 = phase1.groupBy(group_key) \
                   .agg(F.sum('partial_sum').alias('total_sum'),
                        F.sum('partial_count').alias('total_count'))

    return phase2

场景5：窗口函数倾斜

# 方案：限制窗口范围 + 增加分区数
result = df.withColumn(
    'row_num',
    F.row_number().over(
        Window.partitionBy('user_id')
              .orderBy(F.desc('amount'))
              .rowsBetween(Window.unboundedPreceding, Window.currentRow)
    )
).repartition(200)  # 增加分区数

场景6：读写小文件倾斜

# 方案：合并小文件
# 读取时合并
spark.conf.set('spark.sql.files.maxPartitionBytes', 128 * 1024 * 1024)  # 128MB
spark.conf.set('spark.sql.files.openCostInBytes', 4 * 1024 * 1024)      # 4MB

# 写入时合并（Coalesce）
df.coalesce(10).write.parquet('/output/')

# 或使用 AQE 自动合并
spark.conf.set('spark.sql.adaptive.coalescePartitions.enabled', 'true')
spark.conf.set('spark.sql.adaptive.coalescePartitions.minPartitionSize', '1MB')

场景7：笛卡尔积倾斜

# 方案：避免笛卡尔积，改用 Broadcast Nested Loop Join
# 错误写法（笛卡尔积）
df1.crossJoin(df2)

# 正确写法：如果 df2 很小
df1.join(broadcast(df2), how='cross')

# 如果必须笛卡尔积，增加并行度
spark.conf.set('spark.sql.crossJoin.enabled', 'true')
spark.conf.set('spark.sql.shuffle.partitions', 500)

四、Shuffle 优化

4.1 Shuffle 分区数调优

# 黄金公式：shuffle 分区数 = 总数据量 / 目标分区大小
# 目标分区大小：128MB ~ 256MB

total_data_size_mb = 100 * 1024  # 100GB
target_partition_mb = 200         # 200MB
optimal_partitions = total_data_size_mb / target_partition_mb  # 512

spark.conf.set('spark.sql.shuffle.partitions', optimal_partitions)

4.2 AQE 自适应查询执行

# Spark 3.x AQE 三大核心优化
spark.conf.set('spark.sql.adaptive.enabled', 'true')

# 1. 自动合并小分区
spark.conf.set('spark.sql.adaptive.coalescePartitions.enabled', 'true')
spark.conf.set('spark.sql.adaptive.coalescePartitions.minPartitionNum', 1)
spark.conf.set('spark.sql.adaptive.advisoryPartitionSizeInBytes', '128MB')

# 2. 自动切换 Join 策略（SortMerge → Broadcast）
spark.conf.set('spark.sql.adaptive.autoBroadcastJoinThreshold', '20MB')

# 3. 自动处理倾斜 Join
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')

4.3 动态分区裁剪（DPP）

# Spark 3.x 默认开启，无需额外配置
# 效果：大表 join 维度表时，自动裁剪大表分区

# 示例：orders (按日期分区) join dim_date (过滤最近7天)
# DPP 自动将 orders 只扫描最近7天的分区
result = spark.table('orders').join(
    spark.table('dim_date').filter("date >= '2026-05-25'"),
    'date_key'
)
# Spark 自动识别：orders 只需扫描 2026-05-25 ~ 2026-06-01 的分区

五、Join 策略选择

# 强制指定 Join 策略
from pyspark.sql.functions import broadcast

# Broadcast Hint
df1.join(broadcast(df2), 'key')

# Shuffle Merge Hint
df1.join(df2.hint('merge'), 'key')

# Shuffle Hash Hint
df1.join(df2.hint('shuffle_hash'), 'key')

# Shuffle Replicate NL Hint（非等值 join）
df1.join(df2.hint('shuffle_replicate_nl'), df1.col > df2.col)

六、Spark UI 性能诊断

6.1 关键指标速查

6.2 诊断脚本

# scripts/tuning_advisor.py
"""从 Spark History Server 拉取指标，自动生成调优建议"""

def analyze_application(app_id):
    """分析 Spark Application 性能瓶颈"""
    app = get_app_metrics(app_id)
    issues = []

    # 检查1：数据倾斜
    for stage in app['stages']:
        task_times = [t['duration'] for t in stage['tasks']]
        max_time = max(task_times)
        median_time = sorted(task_times)[len(task_times)//2]
        if max_time > median_time * 3:
            issues.append({
                'type': 'DATA_SKEW',
                'stage': stage['id'],
                'severity': 'HIGH',
                'detail': f'Max task {max_time}ms vs Median {median_time}ms',
                'fix': '使用加盐或两阶段聚合'
            })

    # 检查2：Shuffle 过大
    total_shuffle = sum(s['shuffle_read'] for s in app['stages'])
    if total_shuffle > 100 * 1024 * 1024 * 1024:  # 100GB
        issues.append({
            'type': 'EXCESSIVE_SHUFFLE',
            'severity': 'MEDIUM',
            'detail': f'Total shuffle: {total_shuffle/1024/1024/1024:.1f}GB',
            'fix': '增加 Broadcast Join、减少不必要的 Shuffle'
        })

    # 检查3：GC 压力
    for executor in app['executors']:
        gc_ratio = executor['gc_time'] / executor['total_time']
        if gc_ratio > 0.1:
            issues.append({
                'type': 'GC_PRESSURE',
                'severity': 'HIGH',
                'detail': f'GC ratio: {gc_ratio:.1%}',
                'fix': '增加 executor memory 或减少 cache'
            })

    return issues

七、内存管理

# 内存分配公式
# Total Memory = spark.executor.memory
#   ├── Reserved Memory (300MB, 固定)
#   ├── User Memory (40% × (Total - 300MB))
#   └── Spark Memory (60% × (Total - 300MB))
#       ├── Storage Memory (50%)
#       └── Execution Memory (50%)

# 推荐配置
spark.conf.set('spark.executor.memory', '8g')
spark.conf.set('spark.executor.memoryOverhead', '2g')  # 堆外内存
spark.conf.set('spark.memory.fraction', 0.6)            # Spark 可用比例
spark.conf.set('spark.memory.storageFraction', 0.5)     # Storage 占比

# 避免 Spill
spark.conf.set('spark.sql.windowExec.buffer.spill.threshold', 4096)

八、生产 Checklist

• [ ] spark.sql.adaptive.enabled = true（Spark 3.x 必开）
• [ ] spark.sql.adaptive.skewJoin.enabled = true
• [ ] spark.sql.shuffle.partitions 根据数据量合理设置
• [ ] 小表 join 使用 broadcast() hint
• [ ] 避免 collect() 大数据到 Driver
• [ ] 避免 groupByKey，优先用 reduceByKey
• [ ] 避免在 UDF 中使用外部连接
• [ ] Kryo 序列化替代 Java 序列化
• [ ] 合理设置 spark.dynamicAllocation.enabled

九、常见问题