谷歌DataFlow &阅读拼图文件

gsutil split -m 100M gs://your-bucket/largefile.csv gs://your-bucket/largefile_

import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions, GoogleCloudOptions

class ReadShardedFiles(beam.DoFn):
    def process(self, element):
        # 处理每个分片文件
        with open(element, 'r') as f:
            for line in f:
                yield line.strip()

def run():
    options = PipelineOptions()
    gcp_options = options.view_as(GoogleCloudOptions)
    gcp_options.project = 'your-project-id'
    gcp_options.region = 'your-region'
    gcp_options.job_name = 'your-job-name'
    gcp_options.staging_location = 'gs://your-bucket/staging'

    with beam.Pipeline(options=options) as p:
        files = p | 'ListFiles' >> beam.io.ReadFromText('gs://your-bucket/largefile_', skip_header_lines=1)
        lines = files | 'ReadFiles' >> beam.ParDo(ReadShardedFiles())
        # 进一步处理lines

if __name__ == '__main__':
    run()

with beam.Pipeline(options=options) as p:
    files = p | 'ListFiles' >> beam.io.ReadFromText('gs://your-bucket/largefile_', skip_header_lines=1)
    lines = files | 'ReadFiles' >> beam.ParDo(ReadShardedFiles())
    results = lines | 'ProcessLines' >> beam.Map(process_line)
    results | 'WriteResults' >> beam.io.WriteToText('gs://your-bucket/output')