我有一個 GCS 存儲桶，我試圖從中讀取大約 20 萬個文件，然后將它們寫入 BigQuery。問題是我無法創建與代碼配合良好的 PCollection。我正在關注本教程以供參考。我有這個代碼：from __future__ import absolute_importimport argparseimport loggingimport osfrom past.builtins import unicodeimport apache_beam as beamfrom apache_beam.io import ReadFromText, ReadAllFromTextfrom apache_beam.io import WriteToTextfrom apache_beam.metrics import Metricsfrom apache_beam.metrics.metric import MetricsFilterfrom apache_beam.options.pipeline_options import PipelineOptionsfrom apache_beam.options.pipeline_options import SetupOptionsfrom google.cloud import storageimport regex as re# storage_client = storage.Client()# bucket = storage_client.get_bucket('mybucket')## blobs = bucket.list_blobs()# l=list(blobs)# x=[y.name for y in l]# c=x[1:]# print(len(c))files = ['gs://mybucket/_chunk1',         'gs://mybucket/_chunk0']class DataIngestion:    """A helper class which contains the logic to translate the file into    a format BigQuery will accept."""    def parse_method(self, string_input):        x="""{}""".format(string_input)        rx = re.compile(r"""\{[^{}]+\}(*SKIP)(*FAIL)|,""")        d = {}        d['name'], d['date'], d['geometry'], d['value0'], d['value1'], d['value2']=rx.split(x)        d['geometry']=d['geometry'].strip('"')        return ddef run(argv=None):    """Main entry point; defines and runs the pipeline."""    data_ingestion = DataIngestion()    p = beam.Pipeline(options=PipelineOptions())    (p    | 'Create PCollection' >> beam.Create(files)    | 'Read from a File' >> beam.io.ReadAllFromText(skip_header_lines=1)    | 'String To BigQuery Row' >> beam.Map(lambda s:    data_ingestion.parse_method(s))    | 'Write to BigQuery' >> beam.io.Write(    beam.io.BigQuerySink(    'mytable',files問題是如果列表只有一個元素，這段代碼就可以完美運行。只要有超過 1 個元素，轉換“String To BigQuery Row”就會出錯并顯示error: nothing to repeat [while running 'String To BigQuery Row']。這可能與正則表達式模塊有關，但我無法弄清楚出了什么問題，因為它在給定 1 個文件時可以完美運行。