Python中Apache Beam使用CDC写入BigQuery的PCollection格式问题

在Python中使用Apache Beam将数据写入BigQuery时，希望启用最新的CDC（变更数据捕获）特性，但无法确定PCollection中对象的正确格式。官方文档仅说明启用use_cdc_writes选项时，Beam Rows需包含record和row_mutation_info属性。输入的PCollection是含多字段的字典类型DataPoint，尝试过添加对应字段、用Row构造等方式仍未解决。

现有写入代码如下：

pcoll_output = (pcoll_input # | DataPoint:Dict[str,Any] |
    | f'{table_name}:Write to BigQuery' >> WriteToBigQuery(
        full_table_id,  # Target table in BigQuery, e.g. `my-project.my_dataset.my_table_cdc`

        # Table creation options
        create_disposition=BigQueryDisposition.CREATE_NEVER,  # The table should already exists
        primary_key=None,  #type:ignore Used only when creating the table. It should not be used here.

        # Write Method Options
        write_disposition=BigQueryDisposition.WRITE_APPEND,  # Used for STREAMING
        schema=table_config.schema.bigquery_schema_string,  # Required for usage of STORAGE_WRITE_API
        method=WriteToBigQuery.Method.STORAGE_WRITE_API,  # MUCH CHEAPER THAN THE OTHER OPTIONS!

        # CDC/Streaming Options
        use_at_least_once=True,  # True is cheaper and faster but might duplicate records.
        use_cdc_writes=True,  # Can only be True if use_at_least_once=True and method=STORAGE_WRITE_API
        #ignore_insert_ids=False,  # True is cheaper and faster but might duplicate records even when using CDC

        # Test/Retry options
        validate=True,  # Validate the data in Apache Beam before writing to BigQuery. Use for testing
        insert_retry_strategy=bigquery_tools.RetryStrategy.RETRY_ON_TRANSIENT_ERROR
    )
)

已尝试的格式化方式：

• 添加包含变更信息的row_mutation_info字段
• 添加包含DataPoint剩余数据的record字段
• 用以下方式构造Row：

datapoint = Row(
    row_mutation_info=Row(
        mutation_type=self.compute_change_type(datapoint),
        change_sequence_number=self.format_sequence_number(datapoint)
    ),
    record=bigquery_tools.beam_row_from_dict(datapoint, self.schema)
)

正确的数据格式与转换方案 #

核心格式要求 #

启用use_cdc_writes后，PCollection中的每个元素必须包含两个顶级结构：

1. record：对应BigQuery表的实际数据行，结构需与目标表schema完全匹配，支持字典或Beam Row对象
2. row_mutation_info：变更元数据结构，必须包含两个必填字段：
   • mutation_type：变更类型，取值为INSERT、UPDATE、DELETE（大小写敏感）
   • change_sequence_number：全局唯一的字符串类型序列号，用于保证变更的顺序性

适配字典类型DataPoint的转换代码 #

方式1：构造字典对象（简洁高效）

直接将原DataPoint字典包装为符合CDC要求的结构：

def format_for_cdc(datapoint: dict[str, Any]) -> dict[str, Any]:
    return {
        "row_mutation_info": {
            "mutation_type": compute_change_type(datapoint),  # 返回"INSERT"/"UPDATE"/"DELETE"
            "change_sequence_number": str(format_sequence_number(datapoint))  # 转成字符串类型
        },
        "record": datapoint  # 原字典需与目标表schema字段完全匹配
    }

# 在Pipeline中添加转换步骤
pcoll_output = (pcoll_input
    | "Format CDC Data" >> Map(format_for_cdc)
    | f'{table_name}:Write to BigQuery' >> WriteToBigQuery(
        # 保留原有配置参数
        full_table_id,
        create_disposition=BigQueryDisposition.CREATE_NEVER,
        write_disposition=BigQueryDisposition.WRITE_APPEND,
        schema=table_config.schema.bigquery_schema_string,
        method=WriteToBigQuery.Method.STORAGE_WRITE_API,
        use_at_least_once=True,
        use_cdc_writes=True,
        validate=True,
        insert_retry_strategy=bigquery_tools.RetryStrategy.RETRY_ON_TRANSIENT_ERROR
    )
)

方式2：构造Beam Row对象（类型安全）

如果需要强类型校验，可转换为Beam Row对象：

from apache_beam.utils.row import Row

def format_for_cdc(datapoint: dict[str, Any]) -> Row:
    return Row(
        row_mutation_info=Row(
            mutation_type=compute_change_type(datapoint),
            change_sequence_number=str(format_sequence_number(datapoint))
        ),
        record=Row(**datapoint)  # 字典转Row，字段名需与schema一致
    )

# Pipeline应用转换
pcoll_output = (pcoll_input
    | "Format CDC Data" >> Map(format_for_cdc)
    | f'{table_name}:Write to BigQuery' >> WriteToBigQuery(
        # 原有配置参数不变
    )
)

关键注意事项 #

• 目标BigQuery表必须已开启CDC功能（创建表时需指定enable_cdc=true）
• mutation_type取值必须严格符合要求，大小写错误会导致写入失败
• change_sequence_number必须是字符串，且全局唯一，确保BigQuery能正确排序变更
• 若使用beam_row_from_dict工具生成record，需确保其输出结构与目标表schema完全匹配，避免字段类型或名称不匹配导致的验证错误

内容的提问来源于stack exchange，提问作者José Fonseca