如何基于fileio.MatchFiles实现GCS海量文件的Dataflow高效并行？

问题 #

我是Apache Beam初学者，GCS存储桶中有约1100万份文件，结构为yyyy/mm/dd，覆盖6年数据。使用以下Dataflow任务处理：

lines = (p
 | "ReadInputData" >> fileio.MatchFiles(file_pattern = 'gs://bucket/**/*')  
 | "FileToBytes" >> fileio.ReadMatches()
 | "Reshuffle" >> beam.Reshuffle()
 | "GetMetada" >> beam.Map(lambda file: post_process(file))
 | "WriteTableToBQ" >> beam.io.WriteToBigQuery(...)
)

post_process()函数会检查文件元数据，对PDF统计页数、对图片计算哈希。当前任务已运行3天，并行效果极差，多数时间仅1个vCPU工作：先长时间单worker遍历文件，再短暂用约25个worker处理，如此循环。

此前测试单月（14.4万文件）任务时，添加beam.Reshuffle()后，耗时从4小时（单worker）降至16分钟（扩至57个worker）。但全量文件匹配（gs://bucket/**/*）时该方法无效。

我使用Apache Beam Python 3.9 SDK 2.40.0，通过容器化Dataflow模板运行。尝试按年份循环匹配文件：

for year in range(2017, 2023):
     list_yyyy_mm.append( p | f"MatchFile {year:04d}" >> fileio.MatchFiles(file_pattern = f"gs://bucket/{year:04d}/**/*"))

files = (
    list_yyyy_mm
    | "Flatten" >> beam.Flatten()
    ...
)

处理耗时缩短至约2小时，但文件列表阶段仅5个worker，耗时30分钟；按年月循环效果更差。请问使用fileio.MatchFiles处理全量文件的最佳实践是什么？是否需要自定义匹配逻辑？

最佳实践方案 #

1. 拆分文件匹配粒度到日层级 #

利用你已有的yyyy/mm/dd文件结构，直接生成所有日级别的精确匹配路径（比如gs://bucket/2017/01/01/**/*），每个路径对应独立的fileio.MatchFiles变换。

核心逻辑是：每个MatchFiles变换会分配独立worker执行，日级拆分能最大化匹配阶段的并行度（6年约2190个日路径，理论上可拉起近2000个worker并行匹配）。

代码示例：

from datetime import date, timedelta

start_date = date(2017, 1, 1)
end_date = date(2022, 12, 31)
delta = timedelta(days=1)

match_pcollections = []
current_date = start_date

while current_date <= end_date:
    date_path = current_date.strftime("%Y/%m/%d")
    pattern = f"gs://bucket/{date_path}/**/*"
    match_pcollections.append(
        p | f"MatchFiles_{date_path}" >> fileio.MatchFiles(pattern)
    )

files = match_pcollections | "FlattenAllMatches" >> beam.Flatten()

2. 调整Reshuffle的位置到读取文件后 #

全量匹配时Reshuffle无效的原因是：宽模式**/*的匹配阶段由单worker执行，输出的文件列表是单个分区，后续Reshuffle无法解决前置的匹配瓶颈。

正确的并行化时机是在ReadMatches之后：文件被读取为字节流后，立即用beam.Reshuffle()强制打散分区，让后续的post_process能被多worker并行处理。

调整后的流水线结构：

lines = (p
 | # 此处接入日级拆分的MatchFiles + Flatten结果
 | "FileToBytes" >> fileio.ReadMatches()
 | "ForceParallelization" >> beam.Reshuffle()  # 移至此处，确保读取后立即并行
 | "GetMetada" >> beam.Map(lambda file: post_process(file))
 | "WriteTableToBQ" >> beam.io.WriteToBigQuery(...)
)

3. 优化Dataflow Worker配置 #

• 开启吞吐量自动缩放：设置--autoscaling_algorithm=THROUGHPUT_BASED，让Dataflow根据负载自动调整worker数量，避免匹配阶段worker不足。
• 提高初始worker数：针对日级拆分的匹配任务，可设置--num_workers=100作为初始值，快速拉起足够worker完成匹配。
• 选用CPU密集型实例：如果post_process包含PDF页数统计、图片哈希计算等CPU密集操作，优先选择n2-highcpu系列实例，提升单worker处理效率。

4. 无需自定义匹配逻辑 #

Beam的fileio.MatchFiles已针对GCS做了路径匹配优化，自定义匹配逻辑（比如直接调用GCS API枚举文件）反而容易触发分页限制、速率限制等问题，不如结合细粒度路径拆分利用内置变换实现并行匹配。

可选方案：预生成文件列表 #

如果文件结构长期稳定，可提前用GCS命令行工具生成全量文件列表：

gsutil ls gs://bucket/**/* > file_list.txt

然后在Beam中读取该文本文件，逐个读取对应文件内容。这种方式可彻底跳过文件匹配阶段的并行瓶颈，但需要维护文件列表的时效性。

内容的提问来源于stack exchange，提问作者Dr. Fabien Tarrade