delta-rs在GCS产生高额费用的原因及优化方案咨询

费用高昂的原因分析 #

1. Delta元数据的重复读取与扫描 #

每次调用write_deltalake时，delta-rs需要读取Delta表的元数据（_delta_log目录下的日志文件）以获取最新快照，同时会扫描目标分区下的现有文件状态。如果PostgreSQL分批次数多、save_data被频繁调用，这些重复的元数据读取、分区文件列表操作会累积成大量Class B操作；读取日志文件的内容则会产生下载流量。

2. 频繁的小批次写入 #

如果PostgreSQL的分批数据量过小，会导致write_deltalake被多次调用，每次调用都要重复执行元数据校验、分区扫描流程，进一步放大Class B操作的数量。同时，小批次写入会生成大量小Parquet文件，后续无论是delta-rs的自动校验还是业务读取，都需要读取更多文件，增加下载流量和Class B操作。

3. 自定义SUCCESS文件的额外开销 #

你提到为每个上传分区保存SUCCESS文件，这不仅增加了Class A写入操作，还会让delta-rs在扫描分区文件时需要处理更多对象，额外产生Class B操作。Delta Lake本身通过_delta_log保证数据一致性，这类自定义文件完全多余。

4. 元数据缓存缺失 #

默认情况下，直接调用write_deltalake不会复用元数据缓存，每次写入都要重新拉取整个_delta_log的内容和文件列表，尤其是当分区数（25000个）和日志文件较多时，会产生大量下载流量和Class B操作。

代码优化方案 #

1. 合并小批次，减少write_deltalake调用次数 #

将多个PostgreSQL小批次合并为符合max_rows_per_file的大批次后再写入，避免重复执行元数据校验流程：

def save_data(self, df: Generator[pa.RecordBatch, Any, None]):
    batch_buffer = []
    total_rows = 0
    for batch in df:
        batch_buffer.append(batch)
        total_rows += batch.num_rows
        # 累积到目标行数后合并写入
        if total_rows >= self.max_rows_per_file:
            combined_batch = pa.concat_tables(batch_buffer)
            write_deltalake(
                f"gs://<my-bucket-name>",
                combined_batch,
                schema=df_schema,
                partition_by="my_id",
                mode="append",
                max_rows_per_file=self.max_rows_per_file,
                max_rows_per_group=self.max_rows_per_file,
                min_rows_per_group=int(self.max_rows_per_file / 2)
            )
            batch_buffer = []
            total_rows = 0
    # 处理剩余的收尾批次
    if batch_buffer:
        combined_batch = pa.concat_tables(batch_buffer)
        write_deltalake(
            f"gs://<my-bucket-name>",
            combined_batch,
            schema=df_schema,
            partition_by="my_id",
            mode="append",
            max_rows_per_file=self.max_rows_per_file,
            max_rows_per_group=self.max_rows_per_file,
            min_rows_per_group=int(self.max_rows_per_file / 2)
        )

2. 使用DeltaTable复用元数据缓存 #

通过初始化DeltaTable实例，复用元数据快照缓存，避免每次写入都重新读取_delta_log：

# 在类的初始化方法中创建DeltaTable实例
def __init__(self, ...):
    self.delta_table = DeltaTable(f"gs://<my-bucket-name>", schema=df_schema)

def save_data(self, df: Generator[pa.RecordBatch, Any, None]):
    # 使用DeltaTable的write方法，复用已缓存的元数据
    self.delta_table.write(
        df,
        mode="append",
        partition_by="my_id",
        max_rows_per_file=self.max_rows_per_file,
        max_rows_per_group=self.max_rows_per_file,
        min_rows_per_group=int(self.max_rows_per_file / 2)
    )

3. 移除自定义SUCCESS文件 #

删除所有手动生成SUCCESS文件的逻辑，依赖Delta Lake的_delta_log来确认写入的一致性。如果需要验证写入结果，可以通过DeltaTable的latest_version()或history()方法检查日志条目。

4. 调优文件大小参数 #

根据单条记录的平均大小，将max_rows_per_file调整为能生成100MB-1GB大小的Parquet文件（GCS上大文件的读写效率更高，能减少文件数量和后续操作开销）。同时确保min_rows_per_group设置合理，避免过早关闭文件。

5. 启用自动合并与优化写入 #

如果使用的delta-rs版本支持，开启auto_compact和optimize_write参数，自动合并小文件，减少后续读取时的文件数量：

write_deltalake(
    f"gs://<my-bucket-name>",
    ...,
    auto_compact=True,
    optimize_write=True
)

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