Spark SQL 中的动态分区插入和动态分区修剪是两个重要的性能优化特性，它们分别针对数据写入和读取场景进行优化。

1. 动态分区插入实现机制

基础语法与示例

-- 动态分区插入：根据数据自动创建分区
INSERT INTO TABLE sales_partitioned 
PARTITION (country, year, month)
SELECT 
    product_id,
    amount,
    sale_date,
    country,    -- 分区列必须出现在SELECT最后
    YEAR(sale_date) as year,
    MONTH(sale_date) as month
FROM sales_source;

-- 混合分区：部分静态+部分动态
INSERT INTO TABLE sales_mixed
PARTITION (country='US', year, month)  -- country静态，year/month动态
SELECT 
    product_id,
    amount,
    sale_date,
    YEAR(sale_date) as year,
    MONTH(sale_date) as month
FROM sales_source 
WHERE country = 'US';

执行流程解析

配置参数优化

// 启用动态分区
spark.conf.set("spark.sql.sources.partitionOverwriteMode", "dynamic")

// 控制动态分区行为
spark.conf.set("hive.exec.dynamic.partition", "true")
spark.conf.set("hive.exec.dynamic.partition.mode", "nonstrict")

// 限制最大分区数避免小文件问题
spark.conf.set("hive.exec.max.dynamic.partitions", "10000")
spark.conf.set("hive.exec.max.dynamic.partitions.pernode", "1000")

// 优化小文件合并
spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", "true")

实际应用场景

-- 场景1：日志数据按时间分区
INSERT INTO TABLE user_logs_partitioned 
PARTITION (dt, hour)
SELECT 
    user_id,
    action,
    url,
    timestamp,
    DATE(timestamp) as dt,
    HOUR(timestamp) as hour
FROM user_logs_stream;

-- 场景2：电商订单按地区分区  
INSERT INTO TABLE orders_partitioned
PARTITION (region, order_date)
SELECT 
    order_id,
    customer_id,
    amount,
    region,
    DATE(order_time) as order_date
FROM orders_staging;

2. 动态分区修剪实现机制

工作原理

动态分区修剪（Dynamic Partition Pruning）在运行时根据查询条件自动过滤不需要扫描的分区。

-- 示例：DPR自动生效的查询
SELECT * FROM sales_partitioned s
JOIN dates d ON s.sale_date = d.date_key
WHERE d.fiscal_year = 2023;

-- 等价于手动分区过滤（但DPR自动完成）：
SELECT * FROM sales_partitioned s
WHERE s.sale_date IN (SELECT date_key FROM dates WHERE fiscal_year = 2023);

DPR触发条件分析

// 启用DPR（Spark 3.0+默认开启）
spark.conf.set("spark.sql.optimizer.dynamicPartitionPruning.enabled", "true")

// 统计信息相关配置
spark.conf.set("spark.sql.statistics.histogram.enabled", "true")
spark.conf.set("spark.sql.cbo.enabled", "true")

DPR执行流程

3. 高级优化技巧

避免DPR失效的场景

-- ❌ DPR可能失效的情况
SELECT * FROM sales s 
JOIN dates d ON SUBSTR(s.sale_date, 1, 7) = d.month_key  -- 函数操作分区键

-- ✅ 优化为DPR友好的写法
SELECT * FROM sales s 
JOIN dates d ON s.sale_date LIKE CONCAT(d.month_key, '%')

-- ❌ 复杂表达式可能阻碍DPR
SELECT * FROM sales s 
JOIN products p ON s.product_id = p.id 
WHERE p.category = 'ELECTRONICS' OR p.price > 1000

-- ✅ 拆分为UNION ALL保持DPR
SELECT * FROM sales s JOIN products p ON s.product_id = p.id 
WHERE p.category = 'ELECTRONICS'
UNION ALL
SELECT * FROM sales s JOIN products p ON s.product_id = p.id 
WHERE p.price > 1000 AND p.category != 'ELECTRONICS'

统计信息优化DPR

// 确保分区表统计信息准确
spark.sql("ANALYZE TABLE sales_partitioned COMPUTE STATISTICS")
spark.sql("ANALYZE TABLE sales_partitioned PARTITION(dt='2023-01') COMPUTE STATISTICS")

// 列级统计信息收集
spark.sql("ANALYZE TABLE sales_partitioned COMPUTE STATISTICS FOR COLUMNS product_id, amount")

4. 性能对比与监控

DPR效果验证

// 查看执行计划确认DPR生效
val df = spark.sql("""
SELECT s.* FROM sales_partitioned s
JOIN date_dim d ON s.dt = d.date_str
WHERE d.year = 2023 AND d.quarter = 'Q1'
""")

df.explain("extended")

// 期望看到的关键信息：
// - DynamicPartitionPruning 字样
// - 分区过滤条件被下推
// - 扫描的文件数量显著减少

性能监控指标

# Spark UI 中关注以下指标：
# - Scan parquet 操作的实际文件扫描数量
# - 每个Stage的输入数据量
# - 任务执行时间分布

# 通过日志监控DPR效果
spark.conf.set("spark.sql.adaptive.logLevel", "INFO")

5. 实战案例研究

案例1：电商数据分析平台

-- 原始查询（无优化）
SELECT 
    p.category,
    COUNT(DISTINCT s.user_id) as unique_users,
    SUM(s.amount) as total_sales
FROM sales s
JOIN products p ON s.product_id = p.id
JOIN dates d ON s.sale_date = d.date_key
WHERE d.year = 2023 
  AND d.month BETWEEN 1 AND 6
  AND p.department = 'ELECTRONICS'
GROUP BY p.category;

-- 优化策略：
-- 1. 确保sales表按sale_date分区
-- 2. 启用DPR自动过滤2023年1-6月分区
-- 3. 使用合适的文件格式（Parquet/ORC）

案例2：日志分析系统

// 动态分区插入优化配置
def optimizePartitionedWrite(df: DataFrame, tableName: String): Unit = {
  df.write
    .format("parquet")
    .option("compression", "snappy")
    .option("parquet.block.size", "134217728")  // 128MB
    .mode("append")
    .insertInto(tableName)  // 使用insertInto而非saveAsTable
}

// 使用示例
val logsDF = spark.read.json("logs/stream/*.json")
optimizePartitionedWrite(logsDF, "logs_partitioned")

6. 常见问题与解决方案

问题1：小文件过多

// 解决方案：写入后合并小文件
spark.sql("""
INSERT OVERWRITE TABLE sales_partitioned 
PARTITION (dt, hour)
SELECT /*+ COALESCE(10) */ *  -- 提示coalesce
FROM sales_partitioned 
WHERE dt = '2023-01-01'
""")

// 或者使用OPTIMIZE命令（Delta Lake）
spark.sql("OPTIMIZE sales_partitioned ZORDER BY (product_id)")

问题2：DPR不生效

// 诊断步骤：
// 1. 检查统计信息
spark.sql("DESCRIBE EXTENDED sales_partitioned").show()

// 2. 验证分区键数据类型匹配
spark.sql("SHOW PARTITIONS sales_partitioned").show()

// 3. 检查Join条件是否直接使用分区键

问题3：动态分区插入性能差

// 优化写入性能
spark.conf.set("spark.sql.sources.partitionOverwriteMode", "dynamic")
spark.conf.set("spark.sql.hive.convertMetastoreParquet", "true")

// 预排序减少小文件
df.repartition(expr("dt"), expr("hour"))
  .sortWithinPartitions("product_id")
  .write.insertInto("sales_partitioned")

通过合理运用动态分区插入和动态分区修剪，可以大幅提升Spark SQL在处理大规模分区表时的性能和效率。