Como tengo un flujo de datos estático en funcionamiento, me gustaría crear una plantilla a partir de este para permitirme reutilizar fácilmente el flujo de datos sin escribir ninguna línea de comando.

Siguiendo elCrear plantillas el tutorial del oficial no proporciona una muestra de salida con plantilla.

Mi flujo de datos termina con un sumidero BigQuery que toma algunos argumentos como la tabla de destino para el almacenamiento. Este parámetro exacto es el que me gustaría poner a disposición en mi plantilla, lo que me permite elegir el almacenamiento de destino después de ejecutar el flujo.

Pero no puedo hacer que esto funcione. A continuación, pego algunos fragmentos de código que podrían ayudar a explicar el problema exacto que tengo.

class CustomOptions(PipelineOptions):
    @classmethod
    def _add_argparse_args(cls, parser):
        parser.add_value_provider_argument(
            '--input',
            default='gs://my-source-bucket/file.json')
        parser.add_value_provider_argument(
            '--table',
            default='my-project-id:some-dataset.some-table')

pipeline_options = PipelineOptions()

pipe = beam.Pipeline(options=pipeline_options)

custom_options = pipeline_options.view_as(CustomOptions)

(...)

# store
processed_pipe | beam.io.Write(BigQuerySink(
    table=custom_options.table.get(),
    schema='a_column:STRING,b_column:STRING,etc_column:STRING',
    create_disposition=BigQueryDisposition.CREATE_IF_NEEDED,
    write_disposition=BigQueryDisposition.WRITE_APPEND
))

Al crear la plantilla, no di ningún parámetro con ella. En una fracción de segundo recibo el siguiente mensaje de error:

apache_beam.error.RuntimeValueProviderError: RuntimeValueProvider(option: table, type: str, default_value: 'my-project-id:some-dataset.some-table').get() not called from a runtime context

Cuando agrego un--table parámetro en la creación de la plantilla, la plantilla se está creando pero el--table el valor del parámetro se codifica en la plantilla y no se anula por ningún valor de plantilla dado paratable más tarde.

Me sale el mismo error cuando reemplacé eltable=custom_options.table.get(), contable=StaticValueProvider(str, custom_options.table.get()).

¿Hay alguien que ya haya creado un flujo de datos con plantillas con parámetros personalizables BigQuerySink? Me encantaría tener algunas pistas sobre esto.