如何在Google Dataflow Apache Beam Python中读取GCS桶的多个JSON文件

首先你当前的循环写法不符合Apache Beam的执行模型：动态生成多个读步骤会导致Pipeline构建逻辑冗余，路径量较大时还会造成DAG过于复杂，运行效率极低。

最优实现方案 #

Beam的ReadFromText原生支持传入路径列表，无需循环逐个创建读取步骤，直接将你获取到的list_files路径列表作为参数传入即可，代码如下：

import apache_beam as beam
from apache_beam.io.textio import ReadFromText

# 省略list_blobs获取list_files的逻辑
with beam.Pipeline() as p:
    # 直接传入路径列表，自动并行读取所有文件
    all_json_lines = p | "Read all GCS JSON files" >> ReadFromText(list_files)
    # 后续直接基于all_json_lines做解析、转换等处理即可

上述写法输出的all_json_lines就是所有JSON文件的行数据组成的PCollection，完全等同于你循环读取后合并的结果，且Beam会自动优化并行读取逻辑，效率远高于手动循环的实现。

可选扩展方案 #

如果你的路径需要先做过滤、校验等动态处理再读取，可以用FileIO的更灵活的实现：

import apache_beam as beam
from apache_beam.io import fileio

with beam.Pipeline() as p:
    all_json_content = (
        p
        | "Inject path list into pipeline" >> beam.Create(list_files)
        | "Filter valid paths (可选)" >> beam.Filter(lambda x: x.endswith(".json"))
        | "Match file resources" >> fileio.MatchAll()
        | "Read file content" >> fileio.ReadMatches()
        | "Decode UTF-8 content" >> beam.Map(lambda file_obj: file_obj.read_utf8())
    )

额外优化提示 #

如果你的所有JSON文件路径满足统一的通配符规则，甚至可以省略掉自己实现的list_blobs函数，直接把通配符路径传入ReadFromText即可，Beam会自动完成GCS文件匹配：

# 示例：匹配目标桶下所有后缀为.json的文件
all_json_lines = p | "Read all JSON by wildcard" >> ReadFromText("gs://你的桶名/*.json")

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