PySpark写入Delta表的高效分区策略与科学计算方法

分区列科学选型逻辑 #

你当前固定选70-300基数列做分区的方法没有结合数据规模、查询模式，很容易出现分区失效、小文件泛滥或者单分区过大的问题，选型按以下优先级判断：

• 第一优先级是查询过滤覆盖率：只有80%以上业务查询都会携带where过滤条件的列才有资格做分区列，比如dt（日期）、region（区域）、biz_line（业务线）这类通用过滤维度，没有高频过滤场景的列哪怕基数完全匹配也不要选，分区裁剪的收益抵不上写入和元数据维护的开销。
• 第二优先级是单分区数据量：不要硬卡基数区间，核心判定标准是单个分区对应的数据量落在1GB-10GB区间。基数过低（比如性别、是否有效这类只有2-3个值的列）会导致单分区数据过大，查询并发能力差；基数过高（比如用户ID、订单号这类百万级以上基数的列）会生成大量分区目录，给Delta元数据操作带来极大压力，还容易产生海量小文件。
• 第三优先级是数据均匀度：选列前要校验分区列各值的数据占比，不要选存在严重倾斜的列（比如单个值占总数据量20%以上），否则会出现部分Task写入极慢、单分区文件大小超标的问题。
• 基数统计优化：你当前用全表count(distinct)统计基数的方式资源消耗太高，直接用近似统计函数approx_count_distinct(列名, 0.05)即可，误差可以控制在5%以内，统计速度比精确count distinct快一个数量级，同时可以顺便统计列的空值率，空值占比超过10%的列不适合做分区列。

repartition适用场景与正确用法 #

很多人写Delta表小文件泛滥，核心原因就是写入前没做正确的重分区：

• 必须做重分区的场景：
  • 写入分区表前：如果不手动重分区，Spark会按照上游Task的分片直接写入，极易出现一个分区下生成几十上百个小文件，或者单个文件大小超过数GB的问题。
  • 上游数据分片极不均匀：比如数据源是大量KB级小文件、Kafka流数据，上游Task数据量差异超过5倍时，必须重分区平衡数据分布。
• 操作选择：
  • 写入分区表时必须用repartition(目标分区数, 分区列)，这是全量shuffle操作，会把相同分区值的数据路由到同一个Task，从根源上避免单个Task写多个分区生成小文件的问题。
  • coalesce仅适合不需要按列打散、仅需减少分区数的场景（比如大表过滤后分区数暴增、又不想做全量shuffle），写入分区表时不要用coalesce，会导致数据倾斜、文件分布不均。
• 避坑：不要直接调用repartition(数字)不指定分区列，这种方式数据随机分布，还是会出现单个Task跨分区写入的问题，解决不了小文件问题。

分区数科学计算方法 #

不要硬套“总大小/128MB”的固定公式，要结合Delta存储特性、集群资源综合计算：

1. 先算总目标文件数：Delta推荐单文件大小为128MB-512MB，不要小于128MB（会导致IO碎片化），也不要大于1GB（会导致查询并发度不足）。总文件数=总数据量/目标单文件大小，比如总数据量1TB，目标单文件256MB，总文件数就是1024/0.25=4096。
2. 再匹配分区列基数：每个分区下的文件数控制在1-10个为最优，也就是 总文件数/分区列基数 ∈ [1,10]。比如上面算出来总文件数4096，如果选的分区列基数是1024，那每个分区下4个文件，完全符合要求；如果分区列基数只有10，那每个分区下会生成409个文件，属于不合理配置，要么更换基数更高的分区列，要么把目标单文件大小上调到1GB，把总文件数降到1024，保证每个分区下文件数在10个左右。
3. 最后匹配集群资源：最终repartition设置的分区数不要超过集群总CPU核心数的2-3倍，比如集群总共有200个CPU核，分区数最高设到600即可，过高的分区数会带来大量Task调度开销，反而拖慢写入速度。

优化后代码示例 #

优化版基数统计代码 #

-- 替换原来的全表精确count distinct，统计效率提升明显
df.createOrReplaceTempView("my_table")
%sql
select 
    approx_count_distinct(column1, 0.05) as column1_distinct_cnt,
    approx_count_distinct(column2, 0.05) as column2_distinct_cnt,
    sum(case when column1 is null then 1 else 0 end)*1.0/count(1) as column1_null_rate
from my_table

优化版写入代码 #

// 提前计算好的参数：示例场景总数据1TB，分区列dt基数1024，目标单文件256MB
val targetRepartitionNum = 4096
val partitionColumn = "dt"

df.repartition(targetRepartitionNum, col(partitionColumn))
  .write
  .format("delta")
  .mode("overwrite")
  .option("overwriteSchema", "true")
  .option("maxRecordsPerFile", 5000000) // 兜底参数，防止单文件过大，按单条记录大小调整即可，500万条约对应256MB-512MB
  .partitionBy(partitionColumn)
  .save(outputPath)

额外优化提示 #

• 如果是动态分区写入场景，加配置.option("partitionOverwriteMode", "dynamic")，避免全表覆盖误删数据。
• 写入完成后可以执行OPTIMIZE 表名 ZORDER BY (高频查询的非分区列)，进一步合并小文件，加速非分区列的查询效率。高基数的查询维度（比如用户ID、订单ID）不要做分区列，适合作为ZORDER列。

内容的提问来源于stack exchange，提问作者Enrique Benito Casado