在Beam管道中以编程方式生成BigQuery模式

import apache_beam as beam
from apache_beam.io.gcp.bigquery import WriteToBigQuery

# 定义BigQuery表的模式
schema = 'column1:STRING,column2:INTEGER,column3:FLOAT'

# 创建一个Beam管道
pipeline = beam.Pipeline()

# 从数据源读取数据
data = pipeline | beam.io.ReadFromText('input.txt')

# 将数据转换为BigQuery行格式
rows = data | beam.Map(lambda x: x.split(',')) | beam.Map(lambda x: {'column1': x[0], 'column2': int(x[1]), 'column3': float(x[2])})

# 将数据写入BigQuery表
rows | WriteToBigQuery(
    table='project_id.dataset.table',
    schema=schema,
    create_disposition=beam.io.BigQueryDisposition.CREATE_IF_NEEDED,
    write_disposition=beam.io.BigQueryDisposition.WRITE_APPEND
)

# 运行管道
pipeline.run()