基于DeepSeek辅助优化的SQL数据倾斜解决方案实践

EXPLAIN EXTENDED
SELECT 
    user_id,
    COUNT(*) AS action_count,
    AVG(duration) AS avg_duration
FROM user_behavior_log
WHERE dt = '20230915'
GROUP BY user_id;

-- 分析用户行为数据分布
SELECT 
    user_id,
    COUNT(*) as record_count,
    PERCENTILE(COUNT(*), 0.5) OVER() as median_count,
    PERCENTILE(COUNT(*), 0.95) OVER() as p95_count,
    PERCENTILE(COUNT(*), 0.99) OVER() as p99_count
FROM user_behavior_log
WHERE dt = '20230915'
GROUP BY user_id
ORDER BY record_count DESC
LIMIT 100;

-- 第一阶段：局部聚合+随机前缀
SELECT 
    CONCAT(CAST(CEIL(RAND() * 59) AS STRING), '_', user_id) as shuffled_user_id,
    COUNT(*) as partial_count,
    SUM(duration) as partial_sum,
    COUNT(duration) as partial_cnt
FROM user_behavior_log
WHERE dt = '20230915'
GROUP BY CONCAT(CAST(CEIL(RAND() * 59) AS STRING), '_', user_id);

-- 第二阶段：全局聚合
SELECT 
    SUBSTR(shuffled_user_id, INSTR(shuffled_user_id, '_') + 1) as user_id,
    SUM(partial_count) as action_count,
    SUM(partial_sum) / SUM(partial_cnt) as avg_duration
FROM temp_partial_result
GROUP BY SUBSTR(shuffled_user_id, INSTR(shuffled_user_id, '_') + 1);

-- 识别倾斜键
WITH skewed_users AS (
    SELECT user_id
    FROM user_behavior_log
    WHERE dt = '20230915'
    GROUP BY user_id
    HAVING COUNT(*) > 100000
),
normal_data AS (
    SELECT l.*
    FROM user_behavior_log l
    LEFT ANTI JOIN skewed_users s ON l.user_id = s.user_id
    WHERE l.dt = '20230915'
),
skewed_data AS (
    SELECT l.*
    FROM user_behavior_log l
    INNER JOIN skewed_users s ON l.user_id = s.user_id
    WHERE l.dt = '20230915'
)

-- 正常数据处理
SELECT 
    user_id,
    COUNT(*) AS action_count,
    AVG(duration) AS avg_duration
FROM normal_data
GROUP BY user_id

UNION ALL

-- 倾斜键特殊处理
SELECT 
    user_id,
    COUNT(*) AS action_count,
    AVG(duration) AS avg_duration
FROM skewed_data
GROUP BY user_id;

// Spark配置优化
val spark = SparkSession.builder()
    .appName("SkewOptimization")
    .config("spark.sql.adaptive.enabled", "true")
    .config("spark.sql.adaptive.skewJoin.enabled", "true")
    .config("spark.sql.adaptive.skewJoin.skewedPartitionFactor", "5")
    .config("spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes", "256MB")
    .config("spark.sql.adaptive.coalescePartitions.enabled", "true")
    .getOrCreate()

问题：Spark SQL作业中存在数据倾斜，某个task处理2.3亿条记录，其他task最多120万条。

相关表结构：
user_behavior_log (
    user_id string,
    duration double,
    action_type string,
    dt string
)

请提供数据倾斜优化建议。

请生成一个处理数据倾斜的Spark SQL查询模板，要求：
1. 使用salting技术分散热点用户
2. 包含两阶段聚合
3. 处理avg聚合函数的正确性

# 基于DeepSeek建议的Spark参数优化
spark-submit \
    --conf spark.sql.adaptive.enabled=true \
    --conf spark.sql.adaptive.coalescePartitions.enabled=true \
    --conf spark.sql.adaptive.skewJoin.enabled=true \
    --conf spark.sql.adaptive.skewJoin.skewedPartitionFactor=5 \
    --conf spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes=256MB \
    --conf spark.sql.shuffle.partitions=200 \
    --conf spark.sql.adaptive.advisoryPartitionSizeInBytes=64MB \
    your_application.jar