使用python将pubsub转换为不带数据流的bigquery

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

# 设置Pub/Sub订阅和BigQuery表的相关信息
project_id = 'your-project-id'
subscription_id = 'your-subscription-id'
dataset_id = 'your-dataset-id'
table_id = 'your-table-id'

# 创建Pub/Sub订阅和BigQuery客户端
subscriber = pubsub_v1.SubscriberClient()
bigquery_client = bigquery.Client()

# 定义Pub/Sub消息处理函数
def process_message(message):
    # 解析消息数据
    data = message.data.decode('utf-8')
    
    # 在此处进行数据转换和处理
    transformed_data = transform_data(data)
    
    # 将转换后的数据插入到BigQuery表中
    table_ref = bigquery_client.dataset(dataset_id).table(table_id)
    table = bigquery_client.get_table(table_ref)
    rows_to_insert = [(transformed_data,)]
    bigquery_client.insert_rows(table, rows_to_insert)
    
    # 确认消息已处理
    message.ack()

# 订阅Pub/Sub消息
subscription_path = subscriber.subscription_path(project_id, subscription_id)
subscriber.subscribe(subscription_path, callback=process_message)

# 持续监听消息
while True:
    time.sleep(1)