Способы использования параметра провайдера значений в Python Apache Beam

Question

Прямо сейчас я просто могу получить значение RunTime внутри класса, используя ParDo, есть ли другой способ использовать параметр времени выполнения, как в моих функциях?

Это код, который я получил прямо сейчас:

class UserOptions(PipelineOptions):
    @classmethod
    def _add_argparse_args(cls, parser):
        parser.add_value_provider_argument('--firestore_document',default='')

def run(argv=None):

    parser = argparse.ArgumentParser()

    pipeline_options = PipelineOptions()

    user_options = pipeline_options.view_as(UserOptions)

    pipeline_options.view_as(SetupOptions).save_main_session = True

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

        rows = (p 
        | 'Create inputs' >> beam.Create(['']) 
        | 'Call Firestore' >> beam.ParDo(
                CallFirestore(user_options.firestore_document)) 
        | 'Read DB2' >> beam.Map(ReadDB2))

Я хочу, чтобы user_options.firestore_document можно было использовать в других функциях без необходимости выполнять ParDo

Answer 1

Единственный способ использовать поставщиков значений - это ParDos и Combines. Невозможно передать провайдер значения при создании, но вы можете определить DoFn, который возвращает провайдер значения, который вы передаете ему в конструкторе:

class OutputValueProviderFn(beam.DoFn):
  def __init__(self, vp):
    self.vp = vp

  def process(self, unused_elm):
    yield self.vp.get()

И в вашем конвейере вы бы сделалиследующее:

user_options = pipeline_options.view_as(UserOptions)

with beam.Pipeline(options=pipeline_options) as p:
  my_value_provided_pcoll = (
      p
      | beam.Create([None])
      | beam.ParDo(OutputValueProviderFn(user_options.firestore_document))

Таким образом, вы не сможете использовать его в Create, поскольку это невозможно, но вы все равно можете получить его в PCollection.