通过在 Google Cloud Dataflow 中使用 Python SDK 推断架构来读取和写入 avro 文件 - Apache Beam

Question

问题：我正在尝试创建一个云数据流管道，该管道使用 Python SDK 从 Google Cloud Storage 读取 Avro 文件，进行一些处理并在 Google Cloud Storage 上写回 Avro 文件。在查看了 Apache Beam 网站上提供的一些示例后，我尝试运行以下代码。我使用了ReadFromAvro 和WriteToAvro 函数。我想要实现的只是读取 Avro 文件并使用 Dataflow 写入相同的 Avro 文件，但它给了我以下警告并且不输出 avro 文件。

警告/错误：

/Library/Frameworks/Python.framework/Versions/2.7/lib/python2.7/site-packages/apache_beam/io/gcp/gcsio.py:121: DeprecationWarning: object() takes no parameters
  super(GcsIO, cls).__new__(cls, storage_client))
INFO:root:Starting the size estimation of the input
INFO:oauth2client.transport:Attempting refresh to obtain initial access_token
INFO:oauth2client.client:Refreshing access_token
INFO:root:Finished the size estimation of the input at 1 files. Estimation took 0.31790304184 seconds
Traceback (most recent call last):
  File "/Users/USER/PycharmProjects/GCP-gcs_to_bq/gcs-bq.py", line 52, in <module>
    run()
  File "/Users/USER/PycharmProjects/GCP-gcs_to_bq/gcs-bq.py", line 47, in run
    records | WriteToAvro(known_args.output)
TypeError: __init__() takes at least 3 arguments (2 given)

代码：

from __future__ import absolute_import

import argparse
import logging

import apache_beam as beam
from apache_beam.io import ReadFromAvro
from apache_beam.io import WriteToAvro
from apache_beam.options.pipeline_options import PipelineOptions
from apache_beam.options.pipeline_options import SetupOptions


def run(argv=None):
    parser = argparse.ArgumentParser()
    parser.add_argument('--input',
                        dest='input',
                        default='gs://BUCKET/000000_0.avro',
                        help='Input file to process.')
    parser.add_argument('--output',
                        dest='output',
                        default='gs://BUCKET/',
                        #required=True,
                        help='Output file to write results to.')
    known_args, pipeline_args = parser.parse_known_args(argv)
    pipeline_args.extend([
        # CHANGE 2/5: (OPTIONAL) Change this to DataflowRunner to
        # run your pipeline on the Google Cloud Dataflow Service.
        '--runner=DataflowRunner',
        # CHANGE 3/5: Your project ID is required in order to run your pipeline on
        # the Google Cloud Dataflow Service.
        '--project=PROJECT_NAME',
        # CHANGE 4/5: Your Google Cloud Storage path is required for staging local
        # files.
        '--staging_location=gs://BUCKET/staging',
        # CHANGE 5/5: Your Google Cloud Storage path is required for temporary
        # files.
        '--temp_location=gs://BUCKET/temp',
        '--job_name=parse-avro',
    ])
    pipeline_options = PipelineOptions(pipeline_args)
    p = beam.Pipeline(options=pipeline_options)

    # Read the avro file[pattern] into a PCollection.
    records = p | ReadFromAvro(known_args.input)
    records | WriteToAvro(known_args.output)


if __name__ == '__main__':
    logging.getLogger().setLevel(logging.INFO)
    run()

编辑：

我尝试将架构添加到 WriteToAvro 函数，但现在它给了我以下错误：

错误：

/usr/local/bin/python /Users/USER/PycharmProjects/GCP-gcs_to_bq/gcs-bq.py
No handlers could be found for logger "oauth2client.contrib.multistore_file"

/Library/Frameworks/Python.framework/Versions/2.7/lib/python2.7/site-packages/apache_beam/coders/typecoders.py:135: UserWarning: Using fallback coder for typehint: <type 'NoneType'>.
  warnings.warn('Using fallback coder for typehint: %r.' % typehint)

架构：

{"fields": [{"default": null, "type": ["null", {"logicalType": "timestamp-millis", "type": "long"}], "name": "_col0"}, {"default": null, "type": ["null", {"logicalType": "char", "type": "string", "maxLength": 1}], "name": "_col1"}, {"default": null, "type": ["null", {"logicalType": "char", "type": "string", "maxLength": 1}], "name": "_col2"}, {"default": null, "type": ["null", {"logicalType": "char", "type": "string", "maxLength": 1}], "name": "_col3"}, {"default": null, "type": ["null", "long"], "name": "_col4"}, {"default": null, "type": ["null", {"logicalType": "char", "type": "string", "maxLength": 1}], "name": "_col5"}, {"default": null, "type": ["null", {"logicalType": "varchar", "type": "string", "maxLength": 10}], "name": "_col6"}, {"default": null, "type": ["null", "double"], "name": "_col7"}, {"default": null, "type": ["null", "long"], "name": "_col8"}, {"default": null, "type": ["null", {"logicalType": "varchar", "type": "string", "maxLength": 6}], "name": "_col9"}, {"default": null, "type": ["null", {"logicalType": "varchar", "type": "string", "maxLength": 6}], "name": "_col10"}], "type": "record", "name": "baseRecord"}

代码：

pipeline_options = PipelineOptions(pipeline_args)
p = beam.Pipeline(options=pipeline_options)

schema = avro.schema.parse(open("avro.avsc", "rb").read())

# Read the avro file[pattern] into a PCollection.
records = p | ReadFromAvro(known_args.input)
records | WriteToAvro(known_args.output, schema=schema)

Answer 1

问题是数据管道实际上没有得到执行。我设法修复它。解决方案是您需要在以下 2 个选项之一中运行光束管道：

选项 1：

p = beam.Pipeline(options=pipeline_options)

schema = avro.schema.parse(open("avro.avsc", "rb").read())

records = p | 'Read from Avro' >> ReadFromAvro(known_args.input)

# Write the file
records | 'Write to Avro' >> WriteToAvro(known_args.output, schema=schema, file_name_suffix='.avro')

# Run the pipeline
result = p.run()
result.wait_until_finish()

选项 2： 使用pythonwith关键字执行管道：

schema = avro.schema.parse(open("avro.avsc", "rb").read())

with beam.Pipeline(options=pipeline_options) as p:
    records = p | ReadFromAvro(known_args.input)
    records | WriteToAvro(known_args.output, schema=schema, file_name_suffix='.avro')

Answer 2

错误表明您的代码没有将所有必需的参数传递给 WriteToAvro() 转换的构造函数 - 实际上它至少需要 2 个参数（文件名前缀和架构），但这段代码只传递了 1 个（文件名前缀)。

WriteToAvro 当前需要架构：它不是可选参数，并且没有避免指定它的解决方法。原因是 Avro 文件通常需要在创建文件之前提前知道架构，所以WriteToAvro 也需要知道架构。

此外，我们不能从ReadFromAvro 返回的集合中明确地推断出架构：假设用户将--input 传递给与具有多个不同架构的Avro 文件匹配的文件模式——这些架构中的哪一个将WriteToAvro 然后具有要使用吗？