我想在管道中做什么:
- 从pub/sub读取(已完成(
- 将此数据转换为词典(已完成(
- 从dict中获取指定键的值(done(
-
从BigQuery运行一个参数化/动态查询,其中where部分应该是这样的:
SELECT field1 FROM Table where field2 = @valueFromP/S
管道
| 'Read from PubSub' >> beam.io.ReadFromPubSub(subscription='')
| 'String to dictionary' >> beam.Map(lambda s:data_ingestion.parse_method(s))
| 'BigQuery' >> <Here is where I'm not sure how to do it>
从BQ读取的正常方式是:
| 'Read' >> beam.io.Read(beam.io.BigQuerySource(
query="SELECT field1 FROM table where field2='string'", use_standard_sql=True))
我读过关于参数化查询的文章,但我不确定这是否适用于apachebeam。
可以使用侧面输入完成吗?
哪种方法最好?
我尝试过的:
def parse_methodBQ(input):
query=''SELECT field1 FROM table WHERE field1='%s' AND field2=True' % (input['field1'])'
return query
class ReadFromBigQuery(beam.PTransform):
def expand(self, pcoll):
return (
pcoll
| 'FormatQuery' >> beam.Map(parse_methodBQ)
| 'Read' >> beam.Map(lambda s: beam.io.Read(beam.io.BigQuerySource(query=s)))
)
with beam.Pipeline(options=pipeline_options) as p:
transform = (p | 'BQ' >> ReadFromBigQuery()
结果(为什么这样?(:
<Read(PTransform) label=[Read]>
正确的结果应该是:
{u'Field1': u'string', u'Field2': Bool}
解决方案
在管道中:
| 'BQ' >> beam.Map(parse_method_BQ))
函数(使用BigQuery 0.25 API进行数据流(
def parse_method_BQ(input):
client = bigquery.Client()
QUERY = 'SELECT field1 FROM table WHERE field1='%s' AND field2=True' % (input['field1'])
client.use_legacy_sql = False
query_job = client.run_async_query(query=QUERY ,job_name='temp-query-job_{}'.format(uuid.uuid4())) # API request
query_job.begin()
while True:
query_job.reload() # Refreshes the state via a GET request.
if query_job.state == 'DONE':
if query_job.error_result:
raise RuntimeError(query_job.errors)
rows = query_job.results().fetch_data()
for row in rows:
if not (row[0] is None):
return input
time.sleep(1)
您可以读取整个表或使用字符串查询。
我知道您将使用parse_methodBQ方法根据需要自定义查询。由于此方法返回查询,因此可以使用BigQuerySource调用它。这些行在字典里。
| 'QueryTable' >> beam.Map(beam.io.BigQuerySource(parse_methodBQ))
# Each row is a dictionary where the keys are the BigQuery columns
| 'Read' >> beam.Map(lambda s: s['data'])
此外,您可以避免自定义查询并使用过滤方法
关于辅助输入,请查看食谱中的这个例子,以便更好地了解如何使用它们。