时间戳

首页 > 解决方案 > Dataflow Flex 模板验证失败，没有给出任何理由

python - Dataflow Flex 模板验证失败，没有给出任何理由

问题描述

我一直在编写数据流管道并且正在使用弹性模板。

我的代码从 avro 读取并处理它没有问题。但是当涉及到 WriteToAvro 或 WriteToText 时，数据流作业会失败，并且看起来它在模板验证时失败了。我完全没有理由这样做。

我已经尝试了很多东西。删除输出文件的参数并将其硬编码。为 WriteToText 切换 WriteToAvro，但它同样失败。

    with beam.Pipeline(options=options) as p:
        read_from_avro = p \
                         | 'ReadFromAvro' >> ReadFromAvro(input_file)

        redact_data = read_from_avro | "RedactData" >> IdentifyRedactData(project, redact_fields)

        redact_data | 'WriteToAvro' >> WriteToAvro(
                        file_path_prefix=output_file,
                        schema=s,
                        codec='deflate',
                        file_name_suffix='.avro')

的输出join_pcollections是一个 pcollection，每个元素都是一个字典。

数据流日志给出了这个：

2021-06-27 09:04:46.728 BST Workflow failed.

2021-06-27 09:04:46.763 BST Cleaning up.

2021-06-27 09:04:46.817 BST Worker pool stopped.

有谁知道发生了什么事。仅供参考，当我删除最后一步并运行“ProcessData”步骤时，一切运行顺利。这是刚刚中断的最后一个写入步骤。

编辑以添加需求文件。

apache-beam==2.29.0
google-cloud==0.34.0
google-cloud-dlp==3.1.0
google-cloud-storage==1.35.0
google-cloud-core==1.4.1
google-cloud-datastore==1.15.0

如果我尝试使用 apache-beam[gcp]==2.29.0，构建会失败，所以我想知道这是否与此有关。

apache-beam[gcp] 2.29.0 depends on google-cloud-dlp<2 and >=0.12.0; extra == "gcp"

标签： pythongoogle-cloud-dataflowdataflow

解决方案

固定的。我认为问题源于未正确配置管道选项。我还根据 flex wordcount 示例更改了管道的运行方式。

    options = PipelineOptions(beam_args)
    options.view_as(SetupOptions).save_main_session = True
    p = beam.Pipeline(options=options)

    project = options.get_all_options().get('project')

    read_from_avro = p \
                     | 'ReadFromAvro' >> ReadFromAvro(input_file)

    redact_data = read_from_avro | "RedactData" >> IdentifyRedactData(project, redact_fields)

    redact_data | 'WriteToAvro' >> WriteToAvro(
                    file_path_prefix=output_file,
                    schema=table_schema,
                    codec='deflate')

    result = p.run()
    result.wait_until_finish()

推荐阅读