数据倾斜和小文件问题同时出现——两者有什么关联？怎么一起解决？

-- 场景：orders表按channel_id分区，某个channel数据量极大
-- 原始分区情况（倾斜）：
-- partition dt=2024-08-01/channel=A: 500GB（倾斜，1个大分区）
-- partition dt=2024-08-01/channel=B: 1GB
-- partition dt=2024-08-01/channel=C: 500MB
 
-- 加盐打散后的写入（用salt=0-9打散channel=A）：
INSERT INTO orders_salted
SELECT
  order_id, amount,
  CONCAT(CAST(FLOOR(RAND()*10) AS STRING), '_', channel_id) AS salted_channel
FROM orders;
 
-- 打散后的分区情况（小文件问题）：
-- partition salted_channel=0_A: 50GB（500GB/10）
-- partition salted_channel=1_A: 50GB
-- ...（10个分区）
-- partition salted_channel=0_B: 100MB（1GB/10，每个100MB）
-- partition salted_channel=1_B: 100MB
-- ...（10个分区）
-- channel=B原来1个分区变成10个，每个100MB
-- channel=C原来1个分区变成10个，每个50MB
-- 小文件数量增加了10倍！

-- 方案：只打散热点key，非热点key正常写入，最后合并
 
-- 第一步：识别热点key
-- （假设已知channel_id='A'是热点，数据量是其他的500倍）
 
-- 第二步：热点key打散写入，非热点key正常写入
-- 热点key：加盐后写入，避免单分区过大
INSERT INTO orders_processed PARTITION(channel_id)
SELECT
  order_id,
  amount,
  CONCAT('A_salt_', CAST(FLOOR(RAND()*10) AS STRING)) AS channel_id  -- A打散成10份
FROM orders
WHERE channel_id = 'A';
 
-- 非热点key：正常写入，不打散
INSERT INTO orders_processed PARTITION(channel_id)
SELECT order_id, amount, channel_id
FROM orders
WHERE channel_id != 'A';
 
-- 第三步：对打散后产生的小文件做Compaction（以Spark为例）
-- 读取分区数据，控制每个文件128MB左右

# Spark合并小文件：读取分区 → 控制输出文件数 → 重写
from pyspark.sql import SparkSession
 
spark = SparkSession.builder.appName('compaction').getOrCreate()
 
# 设置每个文件目标大小（128MB）
# spark.sql.files.maxRecordsPerFile 或者 repartition控制输出文件数
 
def compact_partition(table_path, partition_spec, target_file_size_mb=128):
    df = spark.read.parquet(f'{table_path}/{partition_spec}')
 
    # 估算输出文件数：总数据量 / 目标文件大小
    total_bytes = df.rdd.map(lambda r: len(str(r))).sum()
    target_files = max(1, int(total_bytes / (target_file_size_mb * 1024 * 1024)))
 
    # 重分区并覆盖写入
    df.coalesce(target_files)  \ # coalesce减少分区数（只减不增）
      .write.mode('overwrite') \
      .parquet(f'{table_path}/{partition_spec}')
 
    print(f'Compacted to {target_files} files')
 
# 对小文件多的分区执行合并
compact_partition('/data/orders_processed', 'channel_id=A_salt_0')
compact_partition('/data/orders_processed', 'channel_id=A_salt_1')
# ...对其他分区执行

从源头同时控制：合理的分区和分桶设计

更好的做法是在数据写入阶段就设计好，让数据既不过于集中也不过于分散：

-- 最佳实践：分区（控制文件数）+ 分桶（控制分布均匀性）组合
 
-- 分区按时间（粗粒度，控制每个分区不要太多文件）
-- 分桶按高基数key（细粒度，保证数据均匀分布）
CREATE TABLE orders_optimized
USING PARQUET
PARTITIONED BY (dt STRING)         -- 按日期分区
CLUSTERED BY (order_id) INTO 200 BUCKETS  -- 按order_id分桶（高基数）
AS SELECT *, date_format(create_time, 'yyyy-MM-dd') AS dt FROM orders;
 
-- 写入时的小文件控制：
spark.conf.set('spark.sql.shuffle.partitions', '200')  -- 和桶数一致
spark.conf.set('spark.sql.files.maxRecordsPerFile', '1000000')  -- 每文件最多100万行
 
-- 查询时的益处：
-- 1. 按dt分区裁剪，只读需要的日期数据
-- 2. 按order_id分桶，数据均匀分布，无倾斜
-- 3. 200桶 × 每桶1个文件 = 200个文件/分区，文件数可控

面试高频追问

追问一：什么大小的文件算小文件？处理小文件的阈值是多少？

没有统一标准，但经验参考：Spark/Hive场景下，每个文件小于128MB（HDFS默认Block大小）就是小文件，理想文件大小是128MB-512MB；Doris/StarRocks场景下，每个Tablet小于100MB就需要关注，理想大小是256MB-1GB。判断小文件是否影响性能：看Task数量（超过文件数的10倍说明文件太多）、看NameNode内存使用率（超过80%需要处理）、看查询规划时间（超过10秒说明文件元数据太多）。处理优先级：先处理影响NameNode稳定性的超大量小文件，再处理影响查询性能的中量小文件。

追问二：Delta Lake/Iceberg有自动处理小文件的能力吗？

有，这是数据湖格式的重要优势之一。Delta Lake的OPTIMIZE命令可以自动合并小文件：OPTIMIZE table_name，会把同一个分区的小文件合并成接近目标大小的大文件（默认1GB）；还可以加上ZORDER BY指定聚簇列，在合并的同时按指定列排序，提升后续查询的数据跳过能力。Iceberg也有类似的rewrite_data_files过程。这些工具让「合并小文件」从手动操作变成了声明式命令，是数据湖相比传统Parquet/ORC文件的重要优势。我在某出海硬件品牌的Databricks环境里，每天凌晨定时跑OPTIMIZE，处理前一天产生的小文件，整体查询性能提升了30%。

追问三：小文件合并会不会产生数据倾斜？

可能会。合并小文件时（比如用coalesce(10)把100个小文件合并成10个），如果这100个小文件本身数据量不均匀（有些100MB，有些1MB），合并后的10个文件仍然大小不一，甚至某个合并后的文件包含了所有大文件，变成一个超大文件。这个超大文件对应的Task在查询时就承担了更多数据，产生新的倾斜。正确的做法是合并时用repartition（而不是coalesce），因为repartition会均匀重新分配数据（通过Shuffle），虽然比coalesce慢，但结果文件大小更均匀。权衡：如果文件原来就大小接近，用coalesce更快；如果原来大小差异很大，用repartition更均匀。

实战踩坑

我在某出海硬件品牌做EMR到Databricks的作业迁移时，遇到了一个典型的「治了倾斜，来了小文件」的场景。有一个广告效果统计作业，原来有严重的channel倾斜（某广告渠道数据量是其他的200倍），我用了加盐打散方案（salt=0-19，20份）处理了倾斜。

-- 加盐处理倾斜后，产生了大量小文件
-- 原来：dt=2024-08-01下有5个channel，每个channel 1-2个文件，共约10个文件
-- 加盐后：5个channel × 20份salt = 100个分区，每个分区2-3个文件 = 200-300个文件
-- 小文件数量增加了30倍，NameNode压力上升，下游作业list文件的时间从1秒到20秒
 
-- 解决方案：加盐处理完倾斜后，立即做Compaction
-- 在Databricks里用OPTIMIZE命令（Delta Lake）
 
-- 写入阶段（加盐处理倾斜）：
spark.sql("""
INSERT INTO delta.`/data/ad_stats_salted`
SELECT
  channel_id,
  FLOOR(RAND() * 20) AS salt,
  SUM(exposure) AS partial_exposure,
  SUM(click) AS partial_click
FROM ad_events
WHERE dt = '2024-08-01'
GROUP BY channel_id, FLOOR(RAND() * 20)  -- 20份局部聚合
""")
 
-- 合并阶段（消除小文件）：
spark.sql("""
OPTIMIZE delta.`/data/ad_stats_salted`
  WHERE dt = '2024-08-01'  -- 只合并今天的分区
""")
-- OPTIMIZE后：200-300个小文件合并成5-8个大文件，查询性能恢复正常

教训：加盐打散倾斜是治倾斜的手术，但手术后要缝合伤口——紧接着做Compaction治小文件。把加盐+Compaction设计成一个完整的流程，而不是两个独立的步骤，才是完整的解决方案。我们后来把这个流程封装成了一个Spark作业模板，每次处理倾斜作业时复用。

总结

倾斜（数据过度集中）和小文件（数据过度分散）是同一个根源的两面。处理倾斜的加盐方案会加剧小文件；暴力合并小文件可能制造倾斜。正确的组合策略：只打散真正倾斜的热点key，打散后立即Compaction；从源头上通过合理的分区+分桶设计，让数据「既不过集中也不过分散」。面试时能把这两个问题的关联和组合处理方案讲清楚，展示了系统性的工程思维。

下一篇聊数据倾斜在面试里的表达策略——用STAR法则怎么把「解决倾斜」的经验讲成一个让面试官记住你的故事？

你们遇到过因为处理倾斜而引入小文件问题的情况吗？最后是怎么解决的？评论区分享～

关注「Wood聊大数据」| 9年大数据老兵，聊技术、聊职场、聊成长