如何用Apache Beam将数据写入多个Google BigQuery表？

1. 从PubSub读取数据 #

利用Apache Beam的PubSubIO组件直接拉取或订阅PubSub的主题/订阅消息，Python示例代码：

import apache_beam as beam
from apache_beam.io import ReadFromPubSub

with beam.Pipeline() as pipeline:
    # 可替换为topic参数指定PubSub主题
    raw_messages = pipeline | "Read from PubSub" >> ReadFromPubSub(subscription="projects/your-project/subscriptions/your-sub")

2. 执行多步Transformations操作 #

根据业务需求串联解析、清洗、字段转换等操作，示例：

import json

def parse_json(message):
    # 解析PubSub消息的JSON内容
    return json.loads(message.decode("utf-8"))

def clean_data(data):
    # 过滤无效数据、补全缺失字段
    data["timestamp"] = data.get("timestamp", "")
    if not data.get("user_id"):
        return None
    return data

def transform_fields(data):
    # 字段重命名、生成衍生字段
    return {
        "user_id": data["user_id"],
        "event_time": data["timestamp"],
        "event_type": data["type"].upper(),
        "event_details": data.get("details", {})
    }

# 串联转换步骤
processed_data = (
    raw_messages
    | "Parse JSON" >> beam.Map(parse_json)
    | "Clean Data" >> beam.Filter(clean_data)
    | "Transform Fields" >> beam.Map(transform_fields)
)

3. 根据配置写入多BigQuery表 #

通过**侧输出（Side Outputs）**实现数据动态路由，适配不同表的写入需求：

3.1 定义表配置与输出标签

from apache_beam.pvalue import TaggedOutput

# 配置：键为输出标签，值为BigQuery表的表名、Schema信息
TABLE_CONFIGS = {
    "user_events": {
        "table": "your-project:dataset.user_events",
        "schema": "user_id:string, event_time:string, event_type:string"
    },
    "error_events": {
        "table": "your-project:dataset.error_events",
        "schema": "user_id:string, error_msg:string, timestamp:string"
    }
}

# 创建对应输出标签
output_tags = {tag: beam.pvalue.TaggedOutput(tag) for tag in TABLE_CONFIGS.keys()}

3.2 数据路由逻辑

class RouteToTable(beam.DoFn):
    def process(self, element):
        # 根据业务规则分发数据，示例：按事件类型区分目标表
        if element["event_type"] in ["LOGIN", "CLICK", "PURCHASE"]:
            yield TaggedOutput("user_events", element)
        else:
            # 构造错误事件格式
            error_data = {
                "user_id": element["user_id"],
                "error_msg": f"Invalid event type: {element['event_type']}",
                "timestamp": element["event_time"]
            }
            yield TaggedOutput("error_events", error_data)

# 执行路由，得到多分支数据
routed_data = processed_data | "Route to Tables" >> beam.ParDo(RouteToTable()).with_outputs(*output_tags.keys())

3.3 写入对应BigQuery表

from apache_beam.io.gcp.bigquery import WriteToBigQuery

# 遍历配置，批量写入各表
for tag, config in TABLE_CONFIGS.items():
    (
        routed_data[tag]
        | f"Write to {tag}" >> WriteToBigQuery(
            table=config["table"],
            schema=config["schema"],
            write_disposition=beam.io.BigQueryDisposition.WRITE_APPEND,
            create_disposition=beam.io.BigQueryDisposition.CREATE_IF_NEEDED
        )
    )

关键注意事项 #

• 若所有表结构一致，可直接通过WriteToBigQuery的table参数接收动态值（从数据中提取表名），简化逻辑。
• 侧输出更适合多表结构不同的场景，能灵活处理分支数据流。
• 生产环境建议增加死信队列，捕获转换或写入失败的数据，避免丢失。

内容的提问来源于stack exchange，提问作者Subham Singhal