如何创建一个以当前日期为输入的beam模板（每天更新）[从获取请求中创建]

Question

我试图创建一个数据流作业每天运行与云调度器。我需要使用get请求从外部API获取数据，所以我需要当前日期作为输入。但是，当我将数据流作业导出为计划模板时，日期输入将停留在执行时，而不是每天更新。我一直在寻找一个解决方案，遇到了ValueProvider，但是我的管道用apache_beam.transforms.create语句总是在没有指定ValueProvider时返回错误'runtimeValueProvider（选项：test,type：str,default_value:'killme').get（）not cally from a runtime context‘。

我能克服这个吗？这似乎是一个如此简单的问题，但我无论如何都不能使它工作。如果有任何想法，我非常感谢！！

Answer 1

您可以使用ValueProvider接口将运行时参数传递给您的管道，要在DoFn中访问它，您需要将其作为参数传入。类似于下面的示例：

https://beam.apache.org/documentation/patterns/pipeline-options/#追溯-logging-runtime-parameters

class LogValueProvidersFn(beam.DoFn):
  def __init__(self, string_vp):
    self.string_vp = string_vp

  # Define the DoFn that logs the ValueProvider value.
  # The DoFn is called when creating the pipeline branch.
  # This example logs the ValueProvider value, but
  # you could store it by pushing it to an external database.
  def process(self, an_int):
    logging.info('The string_value is %s' % self.string_vp.get())
    # Another option (where you don't need to pass the value at all) is:
    logging.info(
        'The string value is %s' %
        RuntimeValueProvider.get_value('string_value', str, ''))

  | beam.Create([None])
  | 'LogValueProvs' >> beam.ParDo(
      LogValueProvidersFn(my_options.string_value)))

您可能还想看看Flex模板：