将多个通配符路径读入 DataFlow PCollection

Question

在我的情况下，我有一堆事件存储在 Storage 的 Date 文件夹下的小文件中。我的数据可能如下所示：

2022-01-01/file1.json
2022-01-01/file2.json
2022-01-01/file3.json
2022-01-01/file4.json
2022-01-01/file5.json

2022-01-02/file6.json
2022-01-02/file7.json
2022-01-02/file8.json

2022-01-03/file9.json
2022-01-03/file10.json

DataFlow 作业将开始和结束日期作为输入，并且需要读取该日期范围内的所有文件。

我正在研究本指南：https://pavankumarkattamuri.medium.com/input-source-reading-patterns-in-google-cloud-dataflow-4c1aeade6831

我看到有一种方法可以将文件列表加载到 PCollection 中：

def run(argv=None):
    # argument parser
    # pipeline options, google_cloud_options
    
    file_list = ['gs://bucket_1/folder_1/file.csv', 'gs://bucket_2/data.csv']
    
    p = beam.Pipeline(options=pipeline_options)
    
    p1 = p | "create PCol from list" >> beam.Create(file_list) \
        | "read files" >> ReadAllFromText() \
        | "transform" >> beam.Map(lambda x: x) \
        | "write to GCS" >> WriteToText('gs://bucket_3/output')

    result = p.run()
    result.wait_until_finish()

我还看到有一种方法可以指定通配符，但我还没有看到它们一起使用。

想知道 beam.Create() 是否支持文件列表中的通配符？这是我的解决方案：

def run(argv=None):
    # argument parser
    # pipeline options, google_cloud_options
    
    file_list = ['gs://bucket_1/2022-01-02/*.json', 'gs://2022-01-03/*.json']
    
    p = beam.Pipeline(options=pipeline_options)
    
    p1 = p | "create PCol from list" >> beam.Create(file_list) \
        | "read files" >> ReadAllFromText() \
        | "transform" >> beam.Map(lambda x: x) \
        | "write to GCS" >> WriteToText('gs://bucket_3/output')

    result = p.run()
    result.wait_until_finish()

还没有尝试过这个，因为我不确定这是否是最好的方法，也没有在网上看到任何类似的例子。想知道我的方向是否正确？

Answer 1

如果无法使用单个通配符，您可以使用以下方法：

def run():
    # argument parser
    # pipeline options, google_cloud_options
 
    with beam.Pipeline(options=pipeline_options) as p:
        file_list = ['gs://bucket_1/2022-01-02/*.json', 'gs://2022-01-03/*.json']

        for i, file in enumerate(file_list):
            p1 = (p
                  | f"Read Text {i}" >> beam.io.textio.ReadFromText(file, skip_header_lines = 0)
                  | "transform" >> beam.Map(lambda x: x)
                  | "write to GCS" >> WriteToText('gs://bucket_3/output'))

我们使用通配符对文件路径执行 foreach。 对于每个元素，我们应用一个具有读写功能的管道。