开源Delta场景下，Spark写入S3生成1GB分区的方案咨询

针对开源Delta无内置Optimize功能的场景，以下几种方法可精准控制写入S3的文件大小在1GB左右：

1. 优化现有分区数计算逻辑（更精准的coalesce/repartition） #

当前仅按总GB数取整的方式易因数据分布不均导致文件大小偏差，可优化为结合总记录数+单条记录平均大小的计算方式：

• 先采样计算单条记录的平均字节大小，再结合总记录数算出总数据量，最后除以1GB（102410241024字节）得到分区数，同时预留10%冗余避免文件超限。
• 代码示例：

import pickle

# 读取Delta表
df = spark.read.format('delta').load(f"s3a://{delta_table}")

# 采样计算单条记录平均大小（采样比例根据数据规模调整）
sample_ratio = 0.01 if df.count() > 1000000 else 0.1
sample_df = df.sample(False, sample_ratio)
avg_record_size = sample_df.rdd.map(lambda x: len(pickle.dumps(x))).mean()

# 计算总数据字节数与目标分区数
total_records = df.count()
total_bytes = total_records * avg_record_size
target_partitions = int(total_bytes // (1024*1024*1024)) + 1 if total_bytes % (1024*1024*1024) !=0 else int(total_bytes // (1024*1024*1024))

# 考虑压缩比（如Snappy压缩比约2:1，需对应调整）
compression_ratio = 2
target_partitions = int(target_partitions / compression_ratio) + 1

# 原分区数小于目标数用repartition（数据分布更均匀），反之用coalesce（无shuffle）
if df.rdd.getNumPartitions() < target_partitions:
    df.repartition(target_partitions).write.format("delta").mode('overwrite').option('overwriteSchema', 'true').save(f"s3a://{delta_table}")
else:
    df.coalesce(target_partitions).write.format("delta").mode('overwrite').option('overwriteSchema', 'true').save(f"s3a://{delta_table}")

2. 使用spark.sql.files.maxRecordsPerFile配置 #

该配置写入时生效，通过控制单文件最大记录数间接控制文件大小：

• 根据单条记录平均大小，计算1GB对应的记录数（例如：1GB=1073741824字节，若单条平均1000字节，则maxRecordsPerFile=1073741）。
• 代码示例：

# 替换为实际采样计算的单条记录平均大小
avg_record_size = 1000
max_records_per_file = int(1073741824 // avg_record_size)
spark.conf.set("spark.sql.files.maxRecordsPerFile", max_records_per_file)

# 直接写入，无需手动指定分区数
df.write.format("delta").mode('overwrite').option('overwriteSchema', 'true').save(f"s3a://{delta_table}")

• 优势：无需提前计算总数据量，适合数据规模动态变化的场景；缺点：若记录大小波动大，文件大小偏差会较明显，适合数据格式均匀的表。

3. 自定义临时分区列实现精准控制 #

对文件大小要求极高的场景，可通过添加临时分区列手动均分数据：

• 计算目标分区数后，添加随机临时列将数据均匀分配到N个分区，写入后移除该列：

from pyspark.sql.functions import floor, rand

target_partitions = 5  # 替换为实际计算的分区数
# 添加临时分区列
df_with_tmp = df.withColumn("tmp_part", floor(rand() * target_partitions))
# 按临时列分区写入
df_with_tmp.write.format("delta").mode('overwrite').option('overwriteSchema', 'true').partitionBy("tmp_part").save(f"s3a://{delta_table}")

# 读取并移除临时列，重新写入（若不需要保留分区列）
final_df = spark.read.format('delta').load(f"s3a://{delta_table}").drop("tmp_part")
final_df.write.format("delta").mode('overwrite').option('overwriteSchema', 'true').save(f"s3a://{delta_table}")

• 注意：Delta表删除列需确保schema兼容，也可后续通过ALTER TABLE DROP COLUMN语句移除临时列。

额外注意事项 #

• 压缩影响：若开启数据压缩（如Snappy、Gzip），实际文件大小会远小于原始数据量，需在分区数计算时加入压缩比系数。
• S3性能：1GB左右的文件大小是S3的最优区间之一，避免单文件过大（如超过5GB）导致读写性能下降。

内容的提问来源于stack exchange，提问作者Michel Senra