使用一个 pcollection 作为另一个 pcollection 的输入

Question

在谷歌数据流中使用python sdk，我想做这样的查询：

query_a_and_b = "从表 A 中选择 a、b"

此查询返回我想用来执行更多查询的元组列表：

query_param = SELECT * from TableA WHERE a = {} and b = {}.format(a, b) （这里我设置了 TableA，但它也将用于与 TableA 内部连接的 TableB、C 和 D...）

所以我想做什么：

coll = (p
    | 'read a_b_tuples' >> beam.io.Read(beam.io.BigQuerySource(query=query_a_and_b, use_standard_sql=True)) 
    | 'Build SQL' >> beam.Map(lambda x: query_param.format(x['a'], x['b'])) 
    | 'Query pardo' >> beam.ParDo(lambda q: [beam.io.Read(beam.io.BigQuerySource(query=q, use_standard_sql=True))])
    | 'Save' >> beam.io.WriteToText('results.csv')
)

我不确定最好的方法，它不起作用。在数据流中实现这一目标的首选方法是什么？

最终，这些查询中的每一个都将返回少量行（少于 5k），我想将它们加载到 pandas 数据帧中进行过滤/处理，然后为每个查询组合所有 TableA、B、C、D tuple (a,b) 并将每个 tuple datafarm 写入一个 csv 文件。

从某种意义上说，我可能错误地减少了问题，我可以使用光束函数按 a 和 b 分组，然后进行处理...？

Answer 1

Beam 尚不直接支持 BigQuery。其他一些转换支持类似的用例，例如JdbcIO.readAll() 可以查询数据库以获取查询参数集合，TextIO.readAll() 可以读取文件名集合 - 但BigQueryIO 还没有这样做，在 Java 和 Python SDK 中都没有。

在您的“查询 pardo”中，您可以改为明确地与 BigQuery REST API 交谈 - 应该没问题，因为您的查询返回少量结果。