使用发布/订阅和数据流从单个JSON创建和插入多行到BigQuery

from google.cloud import pubsub_v1, bigquery
import json

# 初始化Pub/Sub客户端
publisher = pubsub_v1.PublisherClient()
topic_path = publisher.topic_path('your-project-id', 'your-topic-name')

# 初始化BigQuery客户端
client = bigquery.Client()

# 假设我们有一个JSON对象
data = {
    "users": [
        {"id": 1, "name": "Alice", "age": 30},
        {"id": 2, "name": "Bob", "age": 25}
    ]
}

# 将JSON对象转换为字符串并发布到Pub/Sub
future = publisher.publish(topic_path, data=json.dumps(data).encode('utf-8'))

# 确认消息已发布
print(f'Published message ID: {future.result()}')

# 创建一个BigQuery表（如果尚未存在）
dataset_id = 'your_dataset_id'
table_id = 'your_table_id'
table_ref = client.dataset(dataset_id).table(table_id)
schema = [
    bigquery.SchemaField('id', 'INTEGER'),
    bigquery.SchemaField('name', 'STRING'),
    bigquery.SchemaField('age', 'INTEGER')
]
table = bigquery.Table(table_ref, schema=schema)
if not client.get_table(table_ref, timeout=30):
    client.create_table(table)

# 插入数据到BigQuery
rows_to_insert = [(user['id'], user['name'], user['age']) for user in data['users']]
errors = client.insert_rows_json(table, json_rows=rows_to_insert)
if errors:
    print(f'Encountered errors while inserting rows: {errors}')
else:
    print('Rows inserted successfully.')