运行数据流模板作业时出现 RuntimeValueProviderError

Question

试图找出我收到这些错误的原因。快速搜索仅导致提到损坏版本的答案，但这里似乎并非如此。创建模板工作正常，但是当我运行它（并且当我通过限制 arg 时）我得到下面的错误。这个想法是根据模板中提供的参数构建查询。如果有更好的替代方法，我愿意接受。

代码：

class Options(PipelineOptions):
    @classmethod
    def _add_argparse_args(cls, parser):
        parser.add_value_provider_argument(
            '--limit',
            default=0,
            type=int,
            help='Limit the amount of rows retrieved'
        )

...
    
def from_bq(options):
    with beam.Pipeline(options=options) as p:
        (p 
            | 'Read From BQ' >> beam.io.ReadFromBigQuery(query=NestedValueProvider(options.limit, create_query), use_standard_sql=True)
        ) 
    
def create_query(limit):
    query = """
        SELECT * FROM ...
    """

    if limit > 0:
        query = query + " LIMIT {limit}".format(limit=limit)

    return query

错误：

raise error.RuntimeValueProviderError('%s not accessible' % obj)
apache_beam.error.RuntimeValueProviderError: NestedValueProvider(value: RuntimeValueProvider(option: limit, type: int, default_value: 0), translator: create_query) not accessible [while running 'Read From BQ/Read/Split-ptransform-324']

运行 apache-beam 版本 2.27.0。

Answer 1

我认为标准模板无法做到这一点。您应该考虑使用Flex Templates，它具有非模板管道的完全灵活性。