Embedded Debezium如何转SourceRecord为SinkRecord同步MySQL至BigQuery

核心思路是打通ChangeEvent<SourceRecord, SourceRecord>到SinkRecord的转换链路，不需要额外部署重量级组件，就能实现增删改操作、Schema变更的全量同步，有两种可落地的路径：

方案一：轻量嵌入式适配（匹配你当前的Embedded Debezium架构，无需独立Kafka集群） #

这个方案不需要改动你现有Debezium嵌入式监听的逻辑，直接在应用内完成对接：

• 第一步：完成SourceRecord到SinkRecord的转换
  Debezium返回的ChangeEvent携带的SourceRecord本身就包含Kafka Connect运行所需的全部元数据，不需要额外加工字段，直接复用元数据构造SinkRecord即可：

  // 从Debezium变更事件中提取key、value对应的源记录
  SourceRecord keySource = changeEvent.key();
  SourceRecord valueSource = changeEvent.value();
  
  // 构造KCBQ可识别的SinkRecord，参数完全复用源记录的元数据
  SinkRecord sinkRecord = new SinkRecord(
          valueSource.topic(),
          valueSource.kafkaPartition(),
          keySource.keySchema(),
          keySource.key(),
          valueSource.valueSchema(),
          valueSource.value(),
          valueSource.kafkaOffset()
  );
  

  注意不要修改Debezium生成的原始key/value结构：key里默认存储表主键，是KCBQ实现upsert、删除操作的核心依赖；value里自带__op字段标记操作类型（r=全量快照读取、c=新增、u=更新、d=删除），DDL变更会生成携带新Schema定义的独立记录，不需要额外做字段映射。
• 第二步：嵌入式实例化KCBQ Sink组件，跳过独立Kafka Connect部署
  直接在应用内初始化BigQuerySinkTask实例，不需要启动独立的Connect worker进程，必须配置以下核心参数才能支持删改和Schema同步：

  # 开启BQ自动建表、Schema自动演化
  autoCreateTables=true
  autoEvolveSchemas=true
  # 开启删除同步能力
  enableDelete=true
  # 用主键upsert模式写入，避免重复数据
  writeMethod=upsert
  # BQ基础连接配置
  projectId=<你的GCP项目ID>
  defaultDataset=<目标BQ数据集名>
  credentialsFile=<GCP服务账号密钥路径>
  

  日常运行时把转换好的SinkRecord按批次传入sinkTask.put(records)方法，按固定间隔/批次大小调用sinkTask.flush(offsets)触发实际写入，避免数据在内存堆积。
• 第三步：特殊事件处理
  DDL事件转换为SinkRecord后直接传入KCBQ即可，开启autoEvolveSchemas后插件会自动给BQ表加列、适配兼容的字段类型变更；删除事件Debezium会自动生成value为null的tombstone记录，构造SinkRecord时保留null值传入，KCBQ会自动按主键删除BQ中对应行。

方案二：标准生产级架构（稳定性更高，维护成本低） #

如果数据量较大、不想自行维护适配逻辑，可以补一个单节点Kafka（甚至用内嵌Kafka组件，不需要搭集群）：

• 把Embedded Debezium捕获到的SourceRecord直接用Kafka Producer发送到对应topic，topic命名遵循Debezium默认规则连接器名.库名.表名即可
• 启动单节点Kafka Connect进程，同时加载Debezium MySQL Connector、KCBQ插件，不需要做任何类型转换，两个插件原生兼容，增删改、Schema演化逻辑全由官方组件维护，排障和迭代成本更低。

常见踩坑点 #

• Embedded Debezium必须配置持久化的database.history实现，生产环境不要用内存版历史记录存储，否则服务重启后Schema上下文丢失，会导致DDL解析失败、字段错位
• 不要把Debezium生成的结构化记录转成普通JSON字符串传入KCBQ，必须保留原始的Schema信息，否则自动Schema演化、主键识别逻辑会完全失效
• BigQuery本身不支持删除列、字段类型缩容这类不兼容DDL，这类变更触发KCBQ报错时，可以加异常捕获做告警或跳过处理，不要强行修改源数据结构适配

内容的提问来源于stack exchange，提问作者Bhuvan Rawal