这次我们不聊抽象概念，专门看一个数据链路里的实际问题：MySQL 数据同步到 BigQuery，很多团队第一版都是定时批量同步，跑了一段时间后会发现报表对不上、明细缺行、删除和更新根本没同步过去。问题不在写 SQL 的人，而在同步思路本身。要解决这类问题，需要把 MySQL 的 binlog 当成数据源，用 CDC（Change Data Capture）的方式，把数据库的每一次变更准实时地送进 BigQuery。
本文会先拆定期同步漏数据的具体场景，再讲 binlog 的基础格式，然后给出一套可落地的 CDC 架构，包含环境准备、部署步骤、接口/批量任务说明、性能观察和常见问题排查。适合已经在用 BigQuery 做数仓、用 MySQL 做业务库，现在想把同步链路升级成实时或准实时的数据工程师。
1. 核心能力速览
| 能力项 | 说明 |
|---|---|
| 同步方式 | 基于 MySQL binlog 的增量变更捕获，实时推送至 BigQuery |
| 解决的问题 | 删除/更新/物理清理漏同步、批量窗口内数据不一致、报表延迟高 |
| 数据源要求 | MySQL 5.7/8.0 等版本，需开启 binlog，建议使用 ROW 格式 |
| 目标端 | BigQuery 表，天然适合 Append 事件表和按主键 Upsert 的镜像表 |
| 常用方案 | Canal、Debezium + Kafka、Flink CDC、Dataflow 等，可按团队技术栈选择 |
| 部署方式 | Docker、云托管连接器或自建服务，均可对接 |
| 是否支持 API | 支持，CDC 组件通常提供 HTTP/REST 管理接口或通过消息队列对外输出 |
| 是否支持批量任务 | 支持，存量全量导入 + 增量实时同步可组合执行 |
| 延迟水平 | 秒级到分钟级，取决于组件链和消费端写入能力 |
| 适合场景 | 实时数仓、变更追踪、审计分析、主数据同步、缓存/搜索索引联动 |
这里先给结论：如果你的表需要“源端改了目标端马上能看到”，或者需要完整保留更新和删除动作，定时同步不是修修补补能解决的，应该直接切换到 binlog 驱动的 CDC。
2. 为什么定期同步会漏数据
定期同步最常见的形式是：每小时或每天跑一次任务，把 MySQL 里符合条件的行全量拉出来，然后覆盖写进 BigQuery。
这种方案在数据量小、表结构稳定、没有删除操作时能跑，但一旦业务上出现下面几种情况，问题就暴露了。
2.1 同步窗口内的中间变化被覆盖
假设昨天 23:00 跑了一次全量同步，今天 23:00 又要跑第二次。中间 24 小时里，一条订单记录被创建、修改了 5 次，最后又取消。定时任务最终看到的只是“取消”后的最终值，中间 5 次状态流转全部丢失。对于订单、审核、物流这类需要过程分析的场景，这个损失是致命的。
2.2 删除操作无法被完整感知
全量同步的目标表如果是“先清空再写入”或者“按主键覆盖”，你会发现源表物理删除的行，在目标表里可能仍然存在。因为全量同步看到的只是当前快照，它不知道哪些行已经消失。即使做全量比对，也只能在下次对账时发现差异，无法知道这条记录是什么时间、被谁删除的。
2.3 定时批次本身存在延迟窗口
每天一次同步，意味着 BigQuery 里的数据天然滞后 24 小时。就算改成每小时一次，仍然有“批次边界”问题：批处理任务执行到一半时，源库还在发生新变更，这次任务拉到的数据可能不完整，下次任务又可能因为主键重复或时间戳边界和上次重叠，造成重复或缺失。
2.4 失败重跑可能造成重复或脏数据
定时任务经常面临“凌晨跑失败了，早上手动补跑”的情况。补跑如果只按时间窗口重拉，已经写入的数据和新拉到的数据之间没有天然的去重机制。需要在目标表上维护一个“最新更新时间”的判断逻辑，这会让 SQL 越来越复杂，逐步变成一个不可维护的状态。
2.5 DDL 变更会让批量任务直接崩掉
业务表加了一个字段，或者改了字段类型，定时任务的 SELECT * 可能立刻失败，或者字段错位写入 BigQuery。更麻烦的是，批量任务通常不会记录表结构版本，一旦源端 DDL 变化，后续任务很容易持续失败。
这一节的核心结论：定时同步本质上是在对比“两个时刻的快照”，它天然丢失过程、丢失删除、延迟明显，也扛不住 DDL 变化。要拿到连续、完整、可回放的变更流，必须回到 MySQL 内部已经存在的 binlog。
3. binlog 是什么，为什么它适合做同步依据
binlog 是 MySQL 的二进制日志，记录的是数据库层面的所有数据变更逻辑。MySQL 主从复制依赖它，数据恢复也依赖它，而我们做 CDC 同样是消费它。
3.1 binlog 的三种格式
| 格式 | 内容 | 对 CDC 的意义 |
|---|---|---|
| STATEMENT | 记录执行过的 SQL 语句 | 不适合 CDC，因为无法精确还原数据行变化 |
| ROW | 记录每行变更前和变更后的值 | 最适合 CDC，删除、更新、插入都能完整还原 |
| MIXED | 由 MySQL 根据情况自动选择 | 建议不用在 CDC 链路中，可能出现部分语句格式不一致 |
从材料看，绝大多数生产环境在做 CDC 时都会把 binlog_format 设为 ROW，原因很简单：一条 UPDATE 语句在 ROW 格式下会生成多个事件，每个事件包含被修改行的 before 和 after 数据，消费者不需要解析原始 SQL，直接拿到精确值。
3.2 需要关注 binlog 中的事件类型
CDC 链路真正关心的主要事件包括：
- INSERT_ROWS_EVENT：新插入的行数据。
- UPDATE_ROWS_EVENT：更新前后的行数据。
- DELETE_ROWS_EVENT：被删除的行数据。
- TABLE_MAP_EVENT：事件对应的表结构信息。
- QUERY_EVENT / DDL_EVENT：DDL 变更记录，用于感知表结构变化。
- GTID_EVENT：如果开启 GTID，可以拿到全局事务标识，方便精确判断消费位点。
3.3 为什么 binlog 比时间戳字段更可靠
很多人会问：如果表里本来就有 updated_at 字段，能否用“查询 updated_at > 上次同步时间”代替 CDC？
这种做法有两个硬伤。第一，物理删除的行不会保留 updated_at，删除事件直接丢失；第二，如果业务代码更新时没写 updated_at，或者使用了数据库直接修改、存储过程批量更新，时间戳字段可能完全不可信。binlog 是 MySQL 自己生成的，不依赖业务逻辑是否规范，所以可靠性更高。
3.4 GTID 和位点
binlog 消费时，需要记住消费位置。MySQL 提供了两类位置信息：传统的文件名 + 偏移量，以及 GTID 集合。生产级 CDC 工具通常同时支持这两种，推荐开启 GTID。它在故障恢复、链路切换时更稳定，不容易因为 binlog 文件清理或主从切换导致消费位置失效。
4. MySQL 到 BigQuery 的 CDC 整体架构
CDC 链路的基本流程是四个阶段：
MySQL binlog -> 捕获组件 -> 传输组件 -> BigQuery 写入组件捕获阶段读取 binlog 并解析成结构化事件；传输阶段负责把事件送到下游，可能经过消息队列；写入阶段在 BigQuery 里完成追加或合并。
4.1 常用方案对比
| 方案 | 组件 | 特点 | 适合团队 |
|---|---|---|---|
| Canal + 自定义写入 | Canal 解析 binlog，业务服务消费 | 轻量，Java/Scala 技术栈友好，需要自己写下游写入 | 已有 Java 服务，不想引入 Kafka |
| Debezium + Kafka | Debezium Connector + Kafka Connect | 生态成熟，事件格式标准，可接入多个消费端 | 已有 Kafka 基础设施 |
| Flink CDC | Flink SQL / YAML 作业 | 支持 SQL 直接定义同步，批流一体，方便回放和清洗 | 倾向用 Flink 做实时计算 |
| Dataflow + Pub/Sub | 使用 Pub/Sub 接收变更，Dataflow 写 BigQuery | 云上托管，运维少，适合 Google Cloud 深度用户 | 目标就是全托管 |
选型时不用纠结“哪个最好”，而要看团队已有的运行时。如果用 Kafka，接 Debezium 最顺；如果已经在用 Flink 做实时计算，Flink CDC 能少一个转发环节；如果没有实时计算平台，Canal 或轻量脚本更容易落地。
4.2 BigQuery 端模型设计
CDC 到 BigQuery 后，目标表有两种常见模型。
第一种是事件流水表（Append 模型），每个变更事件一行，包含变更前数据、变更后数据、操作类型、binlog 时间戳。适合审计、过程分析和离线回溯。
{ "op": "update", "source_table": "orders", "primary_key_id": "1001", "before": {"status": "pending"}, "after": {"status": "paid"}, "event_time": "2025-01-15 10:30:00" }第二种是镜像表（Upsert 模型），BigQuery 里保留业务表当前最新状态，每次变更事件按主键覆盖或合并。适合 BI 报表直接查询，延迟低，不需要每次跑全量。
实际生产可以两种表都建：流水表用于追溯和重建，镜像表用于日常查询。两张表共用同一条 CDC 链路，只是写入逻辑不同。
5. 环境准备与前置条件
下面给出一条通用的准备清单，具体路径和版本按你的实际环境调整。
5.1 MySQL 侧
确认 MySQL 已开启 binlog，并检查参数：
SHOW VARIABLES LIKE 'log_bin'; SHOW VARIABLES LIKE 'binlog_format'; SHOW VARIABLES LIKE 'binlog_row_image';建议配置：
server_id = 223344 log_bin = mysql-bin binlog_format = ROW binlog_row_image = FULL expire_logs_days = 7 gtid_mode = ON enforce_gtid_consistency = ON其中 binlog_row_image 设置为 FULL，可以保证更新事件里同时有完整的 before 和 after 数据。expire_logs_days 要根据链路消费速度设置，太短会导致下游追不上时 binlog 已被清理，太长会占磁盘。
创建 CDC 专用账号并授权：
CREATE USER 'cdc_user'@'%' IDENTIFIED BY 'your_password'; GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'cdc_user'@'%'; FLUSH PRIVILEGES;需要说明：REPLICATION SLAVE 权限是 Debezium、Canal 等工具读取 binlog 的常用要求，具体看所选工具文档；如果只需要读取业务表数据做全量初始化，保留 SELECT 权限即可。
5.2 BigQuery 侧
提前创建数据集和表。BigQuery 是列式存储，表结构最好按目标业务字段显式定义，不要每次写入时自动推断，避免类型漂移。
命令行创建示例：
bq mk --project_id=your_project your_dataset bq query --use_legacy_sql=false \ 'CREATE TABLE IF NOT EXISTS your_dataset.orders_sink ( order_id STRING, status STRING, amount NUMERIC, updated_at TIMESTAMP, op_type STRING, binlog_ts TIMESTAMP )'如果使用 Storage Write API，需要保证服务账号有bigquery.tables.updateData权限。
5.3 工具部署环境
不管选哪种 CDC 组件，都建议准备 Docker 或独立服务器。下面几条是通用检查项：
- 网络：MySQL 允许 CDC 组件所在主机访问 3306 端口。
- 磁盘：binlog 会占磁盘；CDC 组件所在机器也要预留日志和本地缓冲空间。
- Java：Canal、Debezium、Flink 都依赖 JDK，建议按所选组件要求安装指定 JDK 版本。
- 时间同步：源库、CDC 组件、BigQuery 之间时区不一致，会出现时间偏移，建议统一使用 UTC 并在应用层转换。
6. CDC 链路搭建与运行验证
本节给两条可操作的路线：一条是 Flink CDC 的 SQL/YAML 模式，适合从零快速跑通；另一条是 Debezium + Kafka 的模式，适合已有消息队列的场景。
6.1 快速跑通：Flink CDC 同步到 BigQuery
前提是本机已安装 Flink 或能运行 Flink SQL 客户端，且已经下载对应的 MySQL CDC connector 和 BigQuery connector。以下 YAML 配置是 Flink CDC 常见的本地作业定义，实际连接信息需要替换：
source: type: mysql hostname: 127.0.0.1 port: 3306 username: cdc_user password: your_password tables: app_db.orders server-id: 5400-5404 server-time-zone: Asia/Shanghai scan.incremental.snapshot.enabled: true scan.startup.mode: initial sink: type: bigquery projectId: your_project dataset: your_dataset table: orders_sink writeMethod: STORAGE_WRITE_API启动 Flink CDC 作业后，在 MySQL 里执行几条 INSERT、UPDATE、DELETE，再到 BigQuery 查询目标表，应该能看到事件已经写入。判断成功的标准：
- 新插入的行能在秒级出现在 BigQuery。
- UPDATE 后目标表对应主键的值发生变化。
- DELETE 事件在流水表中出现，或者镜像表里对应行被标记删除。
启动时如果遇到“找不到连接器”的报错，先确认 connector jar 是否放在 Flink 的 lib 目录下。
6.2 生产常用：Debezium + Kafka
Debezium 的 MySQL Connector 以 Kafka Connect 的方式运行。部署时先启动 Kafka 和 Kafka Connect，再提交连接器配置。
连接器配置示例：
{ "name": "mysql-orders-connector", "config": { "connector.class": "io.debezium.connector.mysql.MySqlConnector", "database.hostname": "127.0.0.1", "database.port": "3306", "database.user": "cdc_user", "database.password": "your_password", "database.server.id": "223344", "database.server.name": "mysql-server", "database.include.list": "app_db", "table.include.list": "app_db.orders", "database.history.kafka.bootstrap.servers": "kafka:9092", "database.history.kafka.topic": "schema-changes.app_db", "include.schema.changes": "true", "tombstones.on.delete": "false", "key.converter": "org.apache.kafka.connect.json.JsonConverter", "value.converter": "org.apache.kafka.connect.json.JsonConverter" } }Debezium 输出的 Kafka 消息中，value 里主要字段包括 before、after、source、op。op 的取值一般是 c（新增）、u（更新）、d（删除）、r（初始快照读取）。下游服务消费 Kafka 消息后，再写入 BigQuery。
一种简单的 Python 消费端模板：
import json from google.cloud import bigquery from kafka import KafkaConsumer client = bigquery.Client() table_id = "your_project.your_dataset.orders_sink" consumer = KafkaConsumer( "mysql-server.app_db.orders", bootstrap_servers="127.0.0.1:9092", value_deserializer=lambda m: json.loads(m.decode("utf-8")) ) for message in consumer: event = message.value after = event.get("after") or {} op = event.get("op") row = { "order_id": after.get("order_id"), "status": after.get("status"), "amount": after.get("amount"), "op_type": op, } errors = client.insert_rows_json(table_id, [row]) if errors: print(errors)这段代码只是演示批量消费写入的骨架。生产环境要看情况加上幂等键去重、错误重试、批量攒批等逻辑，避免单条插入导致吞吐过低。
6.3 存量数据初始化
CDC 增量同步只能接住“开启后发生的新变化”，存量数据需要先导入 BigQuery。Flink CDC 的scan.startup.mode: initial会在启动时先做一次全量快照，再继续消费 binlog，这就是“存量 + 增量”的最简单实现。如果自研链路，可以在启动 CDC 前先导出 MySQL 当前数据，再启动增量，最后用这个时间点作为拼接边界。
7. 接口 API 与批量任务设计
7.1 CDC 组件本身的接口
Canal 提供基于 TCP 和 RocketMQ 的客户端接口，也提供 HTTP 管理端口用于查看消费位点；Kafka Connect 提供 REST API，可以提交连接器、查看任务状态。这些接口方便运维，不需要登录服务器看日志。
常用操作示例：
# 查看 Kafka Connect 上的连接器状态 curl http://127.0.0.1:8083/connectors/mysql-orders-connector/status # 重启连接器 curl -X POST http://127.0.0.1:8083/connectors/mysql-orders-connector/restart如果你希望把变更事件直接通过 HTTP 推给下游系统，可以做一个轻量订阅服务，收到 Kafka 消息后调用目标系统 API。这里不涉及具体平台限制，核心是给事件流加一个可控的出口。
7.2 BigQuery 写入方式与批量任务
BigQuery 写入有两条主要路径：
- Storage Write API：适合高吞吐流式写入，支持 exactly-once 语义，是 CDC 推荐的写入方式。
- 批量加载（Load Job）：适合每天把文件导入分区表，CDC 里一般用于存量初始化或周期性对账。
事务性任务设计时要关注“批流关系”。最佳实践是把 CDC 流式写入作为主链路，把每日定时任务作为“校验和补偿”。每日任务扫描 BigQuery 中的镜像表，和 MySQL 做一次主键比对，发现漏掉的变更再补发。这么设计的好处是：流式写入负责低延迟，批量任务负责兜底，不会因为某一条消息丢失造成永久差异。
7.3 批量回放
binlog 天然支持回放。因为事件里记录了变更前后的行数据，只要把这些事件按 gtid 或 binlog 位点排序，再重新写入 BigQuery，就能修复下游表。做回放前注意：
- 事件要按事务边界分组，不要拆散同一个事务的多个事件。
- 回放时目标表需要有主键或去重逻辑。
- 先在一个测试数据集里回放，确认数据量一致后再切换正式环境。
8. 资源占用与性能观察
8.1 MySQL binlog 的额外开销
开启 binlog 本身会有写入开销，ROW 格式下，一条 UPDATE 如果涉及多行，会产生多组 before/after 数据，binlog 文件增长速度会明显加快。观察指标：
- binlog 文件总大小。
- 单位时间 binlog 增量。
- 磁盘剩余空间。
如果 binlog 增长太快、消费端跟不上，排查顺序通常是：先看消费端是否积压，再看是否有大事务一次性修改大量数据，最后确认 binlog 清理策略是否合理。
8.2 CDC 链路的内存与磁盘占用
Canal 和 Debezium 在启动全量快照时，会比较吃内存和临时磁盘，因为要把大规模查询结果做成事件。Flink CDC 做全量快照时也会占用 state。建议给这些组件至少预留 4G 以上内存，特别是在同步 10 万行以上的大表时。
8.3 BigQuery 写入配额
BigQuery 对写入请求数量和单表分区写入有配额限制。CDC 链路如果事件量很大，建议在写入端做攒批，而不是一条一条调用。同时可以把大表按日期分区，避免写入压力都集中到同一个分区。
8.4 性能观察清单
| 观察对象 | 关注指标 | 常见问题 |
|---|---|---|
| MySQL | binlog 增长速率、慢查询 | binlog 被清理导致消费端重连失败 |
| CDC 组件 | 消费延迟、堆内存、线程数 | 大事务造成解析暂停 |
| Kafka | topic 积压量、消费速率 | 下游写入慢造成消息积压 |
| BigQuery | 写入行数、失败请求数 | RPC 超时或限流 |
9. 常见问题与排查方法
| 问题现象 | 可能原因 | 排查方式 | 解决方案 |
|---|---|---|---|
| CDC 启动后立即报“binlog not enabled” | MySQL 未开启 binlog | 执行 SHOW VARIABLES LIKE 'log_bin' | 开启 binlog 并重启 MySQL |
| 停止一段时间后重启，无法找到 binlog 位点 | binlog 被 expire_logs_days 清理 | 查看 MySQL 当前 binlog 文件列表 | 缩短消费停顿时间，或先做全量初始化再增量 |
| Flink CDC 运行时提示 server-id 冲突 | 多个进程使用相同 server_id | 查看连接器配置 | 给每个作业配置不同 server-id 范围 |
| BigQuery 中时间比源库早/晚 8 小时 | 时区处理不一致 | 检查 JDBC 连接时区、Flink 时区参数 | 统一使用 UTC 或明确设置 server-time-zone |
| MySQL 大事务执行后，事件延迟暴增 | 单个事务修改大量行，CDC 需逐个事件解析 | 查看 binlog 中该事务事件数量 | 拆分大事务，或提高 CDC 并发 |
| 删除事件在 BigQuery 里看不到 | 表模型是覆盖式，删除未映射 | 检查 op 为 d 的事件是否被过滤 | 在目标表增加 op_type 字段，或使用标记删除 |
| 写入 BigQuery 出现 quota exceeded | 单分区写入次数过高 | 查看 BigQuery 监控 | 按主键分布分区，写入端攒批 |
这一节是实操中最高频的几个坑。遇到问题第一步永远是看日志，不要直接改配置。Canal、Flink CDC、Debezium 的日志里都会输出解析到哪个 binlog 位点，这个信息对整个链路排错最有价值。
10. 最佳实践与合规建议
10.1 工程化建议
- 第一次跑通时先用小表，不要上来就同步几百 GB 的大表。
- 把 binlog 消费位点持久化。不管是 Kanel、Canal 还是 Flink，消费位点丢失会导致重复或缺失。
- 给 CDC 组件单独建账号，权限限制在业务库的 SELECT 和复制权限，不要用 root。
- 对目标表做分区设计。BigQuery 按时间分区可以降低查询成本和单分区写入压力。
- 对事件消息做幂等键。例如用
source_table + primary_key + binlog_position生成唯一 ID，防止重复消息造成目标表数据重复。 - 保留一段时间原始事件。不要只把字段拆开写入目标表，建议同时保留完整 JSON 事件到另一张表，便于排查数据质量问题。
- 批量补数任务和 CDC 主链路并行时，要设计好重叠窗口，避免同一行被两套逻辑写入。
10.2 数据合规与安全边界
CDC 会把 MySQL 里几乎所有数据变更都搬到 BigQuery，链路一旦打通，数据生命周期就变长了。使用时必须注意：
- 明确同步范围，只同步业务需要的表，不要默认“把所有库同步过去”。
- 涉及用户身份证、手机号、地址等敏感字段时，在链路中做脱敏或掩码处理。
- 授权边界要落到具体账号、数据集、表级别。
- 对增量事件和审计日志设置合理的保留周期。
- 如果同步的是第三方授权数据或公版素材，先确认授权范围允许复制和存储。
- 在生产环境上线前，对同步链路做一次最小权限测试和隐私影响评估。
CDC 不是把数据“多复制一份”这么简单，它实际上是把数据库内部操作永久记录下来。因此，合规策略必须和同步链路同时设计，而不是等功能上线后再补。
11. 小结与下一步
MySQL CDC 到 BigQuery 最核心的转变，是把“定时全量拉取”换成“binlog 事件流持续消费”。它解决的不只是延迟问题，更重要的是把删除、更新、过程变化和 DDL 行为完整地保留下来。对需要实时报表、数据审计、跨系统一致性的团队来说，这条链路值得尽早搭建。
如果今天只做一件事，先把 MySQL 的 binlog 配置检查一遍，再用 Flink CDC 或 Canal 跑通一张小表的增量同步，观察 BigQuery 里 insert、update、delete 三类事件是否都正确反映。跑通后再考虑大表、多表、分布式扩展和批量补偿。最容易踩的坑还是 binlog 清理时机、server-id 冲突、时区不统一和 BigQuery 写入配额，建议收藏备用，等真上线时逐项核对。
