使用Python将发布/订阅消息加载到BigQuery

from google.cloud import bigquery
from google.cloud import pubsub_v1

project_id = 'your-project-id'
topic_name = 'your-topic-name'
message = 'your-message'

publisher = pubsub_v1.PublisherClient()
topic_path = publisher.topic_path(project_id, topic_name)
future = publisher.publish(topic_path, message.encode('utf-8'))

subscription_name = 'your-subscription-name'

subscriber = pubsub_v1.SubscriberClient()
subscription_path = subscriber.subscription_path(project_id, subscription_name)

def callback(message):
    # 处理接收到的消息
    print(f'Received message: {message.data.decode("utf-8")}')
    message.ack()

subscriber.subscribe(subscription_path, callback=callback)

dataset_id = 'your-dataset-id'
table_name = 'your-table-name'

client = bigquery.Client()
dataset_ref = client.dataset(dataset_id)
table_ref = dataset_ref.table(table_name)

job_config = bigquery.LoadJobConfig()
job_config.source_format = bigquery.SourceFormat.DATASTORE_BACKUP
job_config.write_disposition = bigquery.WriteDisposition.WRITE_APPEND

load_job = client.load_table_from_json(
    messages,
    table_ref,
    job_config=job_config
)
load_job.result()

print(f'Loaded {load_job.output_rows} rows into {table_ref.path}')