如何搭建 GCP/Apache Beam 数据流模板？

Question

好的，我必须在这里遗漏一些东西。我需要什么来将管道作为模板？当我尝试通过 these instructions 暂存我的模板时，它运行是模块但没有暂存任何东西。它似乎按预期运行而没有错误，但实际上我没有看到任何文件在我的 --template_location 中添加到存储桶位置收听。我的 python 代码应该出现在那里吗？我假设是这样吧？我已经确定我已经安装了所有的 beam 和 google 云 SDK，但也许我遗漏了什么？您需要做什么来暂存此数据流模板？我也可以手动将文件放入存储桶中并从那里运行吗？以下是我目前正在玩的模板：

import json
import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions
from apache_beam.io.gcp.bigquery import parse_table_schema_from_json

GC_PROJECT = 'my-proj'
BUCKET = 'test-bucket'
STAGING_BUCKET = '%s/test' % BUCKET
TEMP_BUCKET = '%s/test' % BUCKET
# RUNNER = 'DataflowRunner'
RUNNER = 'DirectRunner'

# pipeline_args = ['--save_main_session']
pipeline_args = []
pipeline_args.append('--project=%s' % GC_PROJECT)
pipeline_args.append('--runner=%s' % RUNNER)
pipeline_args.append('--staging_location=gs://%s' % STAGING_BUCKET)
pipeline_args.append('--temp_location=gs://%s' % TEMP_BUCKET)

BQ_DATASET = 'lake'
BQ_TABLE = 'whatever'

SCHEMA_OBJ = [
    {"name": "id", "type": "STRING", "description": ""},
    {"name": "value", "type": "STRING", "description": ""}
]


class ContactUploadOptions(PipelineOptions):
    @classmethod
    def _add_argparse_args(cls, parser):
        parser.add_value_provider_argument(
            '--infile',
            type=str,
            help='path of input file',
            default='gs://%s/data_files/test.csv' % BUCKET)

def run(argv=None):
    print('running')
    p = beam.Pipeline(options=PipelineOptions(pipeline_args))
    lines = (p
             | beam.Create([
                {"id": "some random name", "value": "i dont know"},
                {"id": "id2", "value": "whatever man"}]))

    schema_str = '{"fields": ' + json.dumps(SCHEMA_OBJ) + '}'
    schema = parse_table_schema_from_json(schema_str)
    output_destination = '%s.%s' % (BQ_DATASET, BQ_TABLE)
    (lines
        | 'Write lines to BigQuery' >> beam.io.WriteToBigQuery(
            output_destination,
            schema=schema,
            create_disposition=beam.io.BigQueryDisposition.CREATE_IF_NEEDED,
            write_disposition=beam.io.BigQueryDisposition.WRITE_APPEND))

    p.run().wait_until_finish()


if __name__ == '__main__':
    run(pipeline_args)

此外，如果有人可以 link 一些 sdk documentaion/resources 来解释 how/why 上面的暂存说明应该有效，那就太棒了！

Answer 1

临时位置是在运行作业时加载临时文件的位置。您没有提到 "template_location" 将创建模板的位置。

creating template and running template

请参阅link

如何搭建 GCP/Apache Beam 数据流模板？

How to I stage a GCP/Apache Beam Dataflow template?

python

google-cloud-platform

google-cloud-dataflow

apache-beam