在 PipelineOptions 中传递额外参数时，数据流管道工作人员会停止

Question

我有一个在 Apache Beam 中定义的数据流作业，它正常工作但当我尝试将所有自定义命令行选项包含在我传递给 beam.Pipeline(options=pipeline_options) 的 PipelineOptions 中时中断。它在构建图之后，但在第一步开始之前失败，因为工作人员在启动后变得无响应，最终作业超时，没有有用的日志。

我想传递我的自定义选项，因为只有直接传递给管道的选项才会显示在 Dataflow 控制台 UI 的右侧，并且能够非常方便地查看它们。

全broken example is here。有效的旧版本看起来或多或少像这样

def run():
    parser = argparse.ArgumentParser()
    # Many parser.add_argument lines

    known_args, pipeline_args = parser.parse_known_args()
    pipeline_options = PipelineOptions(pipeline_args)
    pipeline_options.view_as(SetupOptions).save_main_session = True

    with beam.Pipeline(options=pipeline_options) as p:
        # Pipeline definition

不起作用的代码看起来像这样

class CustomOptions(PipelineOptions):
    @classmethod
    def _add_argparse_args(cls, parser):
        # same lines of parser.add_argument

def run():
    pipeline_options = CustomOptions()
    pipeline_options.view_as(SetupOptions).save_main_session = True

    with beam.Pipeline(options=pipeline_options) as p:
        # Same pipeline definition

这是我最终传递给 PipelineOptions 对象的额外键。

api_key
dataset_id
date_column
date_grouping_frequency
input_bigquery_sql
input_mode
org_id
output
output_executable_path # This one isn't really me, it just ends up in there

撇开 argparse/PipelineOptions API 似乎完全基于副作用不谈，我无法理解为什么这会导致作业无法启动。我最好的猜测是我正在通过的选项之一是覆盖/对工作人员产生一些意想不到的副作用，但我以前做过这种事情所以我知道通常可以通过这样的选项和让管道工作。

有人可以发现可能导致第一个工作人员反应迟钝的问题吗？我传递选项的方式似乎是问题所在。

Answer 1

我用你的参数和 Beam 版本 2.41.0 和 Python 3.8.12 进行了测试：

"api_key": "test",
"dataset_id": "test",
"date_column": "test",
"date_grouping_frequency": "test",
"input_bigquery_sql": "test",
"input_mode": "test",
"org_id": "test",
"output": "test",
"output_executable_path": "test"

在 Beam 选项中：

class CustomOptions(PipelineOptions):

    @classmethod
    def _add_argparse_args(cls, parser):
        parser.add_argument("--api_key", help="Api key", required=True)
        parser.add_argument("--dataset_id", help="dataset ID", required=True)

        parser.add_argument("--date_column", help="datdate_column", required=True)
        parser.add_argument("--date_grouping_frequency", help="date_grouping_frequency", required=True)
        parser.add_argument("--input_bigquery_sql", help="input_bigquery_sql", required=True)
        parser.add_argument("--input_mode", help="input_mode", required=True)
        parser.add_argument("--org_id", help="org_id", required=True)
        parser.add_argument("--output", help="output", required=True)
        parser.add_argument("--output_executable_path", help="output_executable_path", required=True)

在 Beam 管道中：

def run():
    custom_pipeline_options = PipelineOptions().view_as(CustomOptions)
    pipeline_options = PipelineOptions()
    
    with beam.Pipeline(options=pipeline_options) as p:
       # Get your custom option arguments
       custom_pipeline_options.api_key
       custom_pipeline_options.dataset_id
       ......

当参数 output_executable_path 是选项的一部分时，出现以下错误：

[2022-11-18, 22:51:38 UTC] 
{beam.py:127} WARNING - argparse.ArgumentError: argument --output_executable_path: conflicting option string: --output_executable_path

Beam 内部使用的参数存在冲突。

当我从选项中删除参数 output_executable_path 时，Dataflow 可以正常工作。

你能在没有这个论点的情况下进行测试吗？