apachebeam从GCS存储桶的多个文件夹中读取多个文件，并加载biquery python

Question

我想每小时设置一个管道，以解析 GCS 存储桶不同文件夹中的 2000 个原始 protobuf 格式文件，并将数据加载到大查询中。到目前为止，我能够成功解析原型数据。

我知道读取文件夹中所有文件的通配符方法，但我现在不想这样做，因为我有来自不同文件夹的数据，我想像并行一样更快地运行，而不是以顺序方式运行

像下面一样

for x,filename enumerate(file_separted_comma):
    --read data from prto
    --load data to bigquery 

现在我想知道以下方法是否是解析 apache beam 中不同文件夹中的多个文件并将数据加载到大查询中的最佳或推荐方法。

还有一件事，从proto解析后的每条记录，我正在将其制作成JSON记录以加载到大查询中，并且不知道这也是将数据加载到大查询而不是直接加载反序列化（解析）原型数据的好方法。

我正在从Hadoop工作转移到数据流，以通过建立这条管道来降低成本。

我是apache-beam新手，不知道什么是缺点

import time
import sys
import argparse
import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions
import csv
import base64
import rtbtracker_log_pb2
from google.protobuf import timestamp_pb2
from google.protobuf.json_format import MessageToDict
from google.protobuf.json_format import MessageToJson
import io
from apache_beam.io.filesystems import FileSystems


def get_deserialized_log(serialized_log):
    log = rtbtracker_log_pb2.RtbTrackerLogProto()
    log.ParseFromString(serialized_log)
    return log


def print_row(message):
    message=message[3]
    message = message.replace('_', '/');
    message = message.replace('*', '=');
    message = message.replace('-', '+');
    #finalbunary=base64.b64decode(message.decode('UTF-8'))
    finalbunary=base64.b64decode(message)
    msg=get_deserialized_log(finalbunary)

    jsonObj = MessageToDict(msg)
    #jsonObj = MessageToJson(msg)
    return jsonObj

def parse_file(element):
  for line in csv.reader([element], quotechar='"', delimiter='\t', quoting=csv.QUOTE_ALL, skipinitialspace=True):
    return line



def run():
    parser = argparse.ArgumentParser()
    parser.add_argument("--input", dest="input", required=False)
    parser.add_argument("--output", dest="output", required=False)
    app_args, pipeline_args = parser. parse_known_args()

    with beam.Pipeline(options=PipelineOptions()) as p:
        input_list=app_args.input
        file_list = input_list.split(",")
        res_list = ["/home/file_{}-00000-of-00001.json".format(i) for i in range(len(file_list))]

        for i,file in enumerate(file_list):
            onesec=p | "Read Text {}".format(i) >> beam.io.textio.ReadFromText(file)
            parsingProtoFile=onesec | 'Parse file{}'.format(i) >> beam.Map(parse_file)
            printFileConetent=parsingProtoFile | 'Print output {}'.format(i) >>beam.Map(print_row)
        
            #i want to load to bigquery here
            ##LOAD DATA TO BIGQUERY

            #secondsec=printFileConetent | "Write TExt {}".format(i) >> ##beam.io.WriteToText("/home/file_{}".format(i),file_name_suffix=".json", 
###num_shards=1 , 
##append_trailing_newlines = True)
        

if __name__ == '__main__':
    run()

在本地运行下面的代码

python3 another_main.py --input=tracker_one.gz,tracker_two.gz

输出路径我没有确定，因为我不想将数据保存到 GCS，因为我会将其加载到 bigquery 中

并且像下面这样在dataflowrunner中运行

python3 final_beam_v1.py --input gs://bucket/folder/2020/12/23/00/00/fileread.gz --output gs://bucket/beamoutput_four/ --runner DataflowRunner --project PROJECT --staging_location gs://bucket/staging_four --temp_location gs://bucket/temp_four --region us-east1 --setup_file ./setup.py --job_name testing

Answer 1

这种读取文件的方法很好（只要输入文件的数量不是太大）。但是，如果您可以将要读取的文件集表示为通配符表达式（可以与多个文件夹匹配），则可能会表现得更好，Dataflow将并行读取所有与模式匹配的文件。

对于写入BigQuery，最好使用内置的BigQuery接收器。默认行为是创建JSON格式的临时文件，然后将其加载到BigQuery中，但也可以改用Avro，这会更有效。您还可以使用Flatten将所有输入组合到一个PCollection中，这样您的管道中只需要一个BigQuery接收器。