使用 ValueProvider 在数据流中格式化 BigQuery

Question

我目前正在使用Dataflow在python中进行循环批处理。

基本上，我从bigquery中读取数据并在其上做一些事情。我的管道如下所示

pipeline_options = PipelineOptions()
p = beam.Pipeline(options=pipeline_options)
lines = (p
| 'read_big_query' >> beam.io.Read(beam.io.BigQuerySource(query=QUERY, use_standard_sql=True))
| "doing stuff" >> beam.Map(do_some_stuff)
)

我想使用数据流模板运行作业以使其适应运行时。

感谢文档 https://cloud.google.com/dataflow/docs/guides/templates/creating-templates，在您的函数部分中使用ValueProvider，我设法使用ParDo从运行时为"do_some_stuff"提供了一个额外的参数。

class TemplateOption(PipelineOptions):
@classmethod
def _add_argparse_args(cls, parser):
parser.add_value_provider_argument('--template_do_stuff_param',
default=45,
type=int)
class MyDoStuffFn(beam.DoFn):
def __init__(self, template_do_stuff_param):
self.template_do_stuff_param = template_do_stuff_param
def process(self, *_):
yield do_some_stuff(self.template_do_stuff_param.get())

pipeline_options = PipelineOptions()
p = beam.Pipeline(options=pipeline_options)
template_option = pipeline_options.view_as(TemplateOption)
lines = (p
| 'read_big_query' >> beam.io.Read(beam.io.BigQuerySource(query=QUERY),
               use_standard_sql=True))
| "doing stuff" >> beam.ParDo(MyDoStuffFn(template_option.template_do_stuff_param))
)

但是我还想更改与进程相关的用户数量，因此我想使查询适应运行时。

class TemplateOption(PipelineOptions):
@classmethod
def _add_argparse_args(cls, parser):
parser.add_value_provider_argument('--template_nb_users',
default=100,
type=int)
parser.add_value_provider_argument('--template_do_stuff_param',
default=45,
type=int)
class MyDoStuffFn(beam.DoFn):
def __init__(self, template_do_stuff_param):
self.template_do_stuff_param = template_do_stuff_param
def process(self, *_):
yield do_some_stuff(self.template_do_stuff_param.get())

pipeline_options = PipelineOptions()
p = beam.Pipeline(options=pipeline_options)
template_option = pipeline_options.view_as(TemplateOption)
lines = (p
| 'read_big_query' >> beam.io.Read(beam.io.BigQuerySource(query=QUERY.format(nb_users=template_option.template_nb_users.get()),
               use_standard_sql=True))
| "doing stuff" >> beam.ParDo(MyDoStuffFn(template_option.template_do_stuff_param))
)

。这不起作用，因为我在管道执行之前调用 get((。到目前为止，我还没有设法将我为do_some_stuff功能所做的调整为"读取"行

任何关于如何进行的建议或解决方案将不胜感激。谢谢！

Answer 1

遗憾的是，BigQuerySource不支持值提供程序。这是因为它是在数据流运行程序中本机实现的，因此所有信息都需要在管道构建时可用。

您可以尝试转换apache_beam.io.gcp.bigquery.ReadFromBigQuery- 这将允许您使用值提供程序。