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

基础概念

发布/订阅模式（Pub/Sub）：这是一种消息传递模式，其中消息的发送者（发布者）不会直接向接收者（订阅者）发送消息。相反，消息被发布到一个主题（Topic），订阅者可以订阅这个主题并接收消息。

数据流（Data Streaming）：指的是实时传输数据的技术，允许数据在产生后立即被传输和处理，而不是批量处理。

BigQuery：是一个完全托管的数据仓库服务，允许用户运行复杂的数据分析查询，并且能够处理大规模数据集。

类型

消息类型：可以是JSON、XML、二进制等格式。
主题类型：可以是有状态的或无状态的，取决于消息处理的需求。

应用场景

实时数据分析：如股票交易分析、用户行为跟踪等。
日志处理：收集和分析来自不同服务的日志数据。
物联网数据处理：处理来自传感器的大量实时数据。

示例代码

以下是一个使用Python和Google Cloud Pub/Sub以及BigQuery的简单示例，展示如何从单个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.')