【发布时间】：2019-03-13 16:31:51
【问题描述】：
我需要读取 BQ 表并应用一些转换并将其加载到另一个 BQ 表。转换对所有表都是通用的。
我只是想知道我们是否可以一次读取多个表并应用转换并加载到不同的目标表。源表和目标表的结构将相同。例如
table_x --transformation(abc) -- table_x1
table_y --transformation(abc) -- table_y1
下面是我用一个表测试的示例代码：
import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions
from apache_beam.io.gcp.bigquery import parse_table_schema_from_json
def get_schema(table_in):
"""
Function to pull the json schema for the table being copied. The schema should be in the schema/ folder, and be of the format:
'schema_"input_table_name".json'
"""
with open('schema/schema_'+table_in+'.json') as f:
data = f.read()
# Wrapping the schema in fields is required for the BigQuery API.
table_schema = '{"fields": ' + data + '}'
return parse_table_schema_from_json(table_schema) # # This code reads from a biq query table and loads in to another table on BQ, transformation still need to looked at
def run(argv=None):
"""
Function that will instantiate the pipeline options, define the pipeline, and then run it
"""
pipeline_options = PipelineOptions()
p = beam.Pipeline(options=pipeline_options)
(p
| 'ReadTable' >> beam.io.Read(beam.io.BigQuerySource('projecttest:datasetx.table_x'))
| 'Write to BigQuery' >> beam.io.Write(
beam.io.BigQuerySink('projecttest:datasetx.table_x1',
schema=get_schema('table_x'),
write_disposition=beam.io.BigQueryDisposition.WRITE_TRUNCATE,
create_disposition=beam.io.BigQueryDisposition.CREATE_IF_NEEDED))
)
p.run().wait_until_finish()
if __name__ == '__main__':
run()
【问题讨论】：
标签： python google-bigquery apache-beam