企业官网建设流程全解析

做数据同步的人，迟早会遇到同一个问题：业务方过来问“为什么数仓里的订单数和线上差那么多”，而你只能回答“下次同步还没跑”“延迟半小时”。这种对话出现得越来越频繁，往往意味着定时同步方案已经撑不住业务体量了。本文围绕 MySQL CDC（Change Data Capture，变更数据捕获）到 BigQuery 的同步链路展开，重点对比周期性同步的盲区，并通过 binlog 讲清楚 CDC 的底层原理，最后给出一套可复现的 Debezium + Kafka 实验环境。

不管你是正在维护数据管道的数据工程师，还是想把业务库数据搬到数仓的后端开发者，都能从这篇文章里找到可以直接落地的思路。文章中不会只讲概念，也不会只贴代码，每一步都会解释“为什么这么做”，以及生产环境里容易踩的坑。

1. 周期性同步到底丢掉了什么

1.1 定时同步的两种常见实现

在引入 CDC 之前，大多数团队的同步方案是“定时跑脚本”。思路通常有两种。

第一种是时间戳增量同步。假设源表里有一个update_time字段，脚本每隔几分钟或几小时执行一次，读取大于上次同步时间点的数据，写入目标库。

# 伪代码：每 5 分钟执行一次 def sync_orders(): last_time = get_last_sync_time() with source_db.cursor() as cur: cur.execute( "SELECT * FROM orders WHERE update_time > %s", last_time ) rows = cur.fetchall() for row in rows: upsert_to_bigquery(row) set_last_sync_time(now())

第二种是全量比对同步。把源表和目标表的数据全部拉下来，逐条比较差异，再统一更新。这种方案在小表阶段很好用，但一旦数据量上去了，每次跑全量都是对源库和目标库的双重压力。

# 伪代码：全量比对 def full_sync_orders(): source_rows = load_all_source_orders() target_rows = load_all_target_orders() diff = compare_two_sets(source_rows, target_rows) apply_diff(diff)

这两种方式看起来都能工作，但它们在数据一致性上存在天然盲区。

1.2 三个最容易翻车的场景

先给出一张典型的订单表结构，后面所有讨论都围绕这张表展开：

CREATE TABLE orders ( id BIGINT PRIMARY KEY AUTO_INCREMENT, order_no VARCHAR(32) NOT NULL, user_id BIGINT NOT NULL, amount DECIMAL(10, 2) NOT NULL, status VARCHAR(20) NOT NULL, create_time DATETIME NOT NULL, update_time DATETIME NOT NULL, KEY idx_status (status) );

场景一：物理删除无法感知。

假如业务方误插入了一批测试订单，或者在风控流程里删除了异常订单，数据库执行的是DELETE FROM orders WHERE id = 1001。时间戳增量同步完全感知不到这次删除，目标库 BigQuery 里那条脏数据会一直存在。有人会说“我们业务不做物理删除，只做软删除”。但实际情况是，总有一些表由于历史原因没有软删除字段，或者 DBA 手动清理数据，这类例外恰恰是数据质量事故的高发区。

场景二：业务字段更新不规范。

时间戳增量同步有一个强依赖：业务代码在 UPDATE 时必须正确维护update_time字段。但现实是，很多 UPDATE 语句并不会主动带上这个字段。

-- 没有更新 update_time，甚至直接改主键 UPDATE orders SET status = 'PAID' WHERE order_no = 'A001';

如果有人执行了上面的语句，或者直接用工具连库修改了几条老数据，增量同步下一次扫描时根本扫不到这些变化。数据不一致就这样悄悄发生了。

场景三：源库压力不可控。

全量比对方案每次都要全表扫描，索引再优化也扛不住每天几十次扫描；大促、月末对账期间，这种同步脚本还会和业务抢数据库资源，导致慢查询变多，甚至拖慢线上接口。对数据量大的表来说，周期同步的时间窗会越来越长，最终从“每 5 分钟一次”退化到“每小时一次”甚至“明天再说”。

2. CDC 是什么，binlog 为什么是关键

2.1 CDC 的核心思想

CDC 全称 Change Data Capture，中文叫“变更数据捕获”。它的核心思想是：不主动去查“哪些数据变了”，而是让数据库主动告诉下游“哪一行数据发生了变化”。

打个比方，周期性同步像保安每隔半小时去仓库清点一遍货，数完才能知道少了什么；而 CDC 相当于在仓库门口装了一个感应器，有人进出就会触发记录。两者获取信息的及时性和准确度完全不同。

实现 CDC 的技术方案有很多种，MySQL 生态里最常见的做法是解析 binlog。

2.2 binlog 是 MySQL 的“操作日志”

binlog 是 MySQL Server 层生成的二进制日志，记录了所有导致数据变化的操作。主从复制就是基于 binlog 实现的：主库把 binlog 发给从库，从库重放这些日志，从而保持数据一致。

binlog 有三种格式，这一点在配置 CDC 时非常关键：

• STATEMENT（语句格式）：记录的是 SQL 原文。比如UPDATE orders SET status='PAID'，数据量小，但主从不一致风险高。
• ROW（行格式）：记录的是每行数据“变化前”和“变化后”的具体内容。信息最完整，最适合 CDC。
• MIXED（混合格式）：默认使用 STATEMENT，遇到不确定操作时自动切换为 ROW。

对于 CDC 工具来说，必须把 binlog 设置为 ROW 格式，并且把binlog_row_image设置为FULL，才能拿到每一行数据变化前后的完整镜像。

2.3 CDC 工具如何“伪装成从库”

MySQL 主从复制的模型是：从库连接主库，请求指定 binlog 位置之后的日志，然后重放。

CDC 工具的工作原理与此类似：工具把自己伪装成一个“从库”，向 MySQL 注册一个唯一的server-id，请求从某个 binlog 位置开始同步。MySQL 把 binlog 事件源源不断地发给它，工具解析这些事件，转换成结构化的消息，再发给下游。

正因为 binlog 是 MySQL 原生日志，CDC 方案对源库业务代码几乎零侵入。不需要业务方改表结构，不需要统一维护update_time，也不需要每张表都做软删除改造。

3. 主流 MySQL CDC 工具怎么选

3.1 Debezium + Kafka Connect

Debezium 是目前社区最活跃的 CDC 框架之一。它基于 Kafka Connect 运行，把 MySQL binlog 事件转换成标准化的 JSON 或 Avro 消息，写入 Kafka。

这种方案的优势很明显：Kafka 作为中间缓冲层，可以解耦源库和目标库；消息可以重复消费，方便多个下游使用同一份变更数据；同时 Kafka 本身具备消息堆积能力，能应对下游短时故障。

3.2 Flink CDC

Flink CDC 是 Flink 生态的连接器，底层同样通过 Debezium 解析 binlog，但把能力封装成了 Flink 的 Source。它最大的优势是能和 Flink SQL、Flink 流处理任务无缝集成，适合需要做实时计算、实时关联的场景。

3.3 Canal

Canal 是阿里巴巴开源的项目，专门针对 MySQL binlog 解析。它的诞生最初是为了解决业务缓存同步问题，因此很多 Java 技术栈的团队用得比较多。Canal 可以将 binlog 发送到 MQ（如 RocketMQ、Kafka）或者直接推送给下游应用。

3.4 选型建议

选型没有唯一答案，主要看现有技术栈：

• 如果团队已经有 Kafka，并且希望下游有多方消费，优先选Debezium + Kafka Connect。
• 如果团队重度使用 Flink，需要边同步边计算，优先选Flink CDC。
• 如果团队是 Java 技术栈，数据链路简单，Canal也完全足够。

本文后续的实战部分以 Debezium 为主，因为它最接近“binlog 订阅与分发”的本质，也方便把每一步拆开看清楚。

4. 环境准备：开启 MySQL binlog

4.1 确认当前 binlog 状态

在开始之前，先确认 MySQL 是否已经开启了 binlog。使用 MySQL 客户端登录后执行：

SHOW VARIABLES LIKE 'log_bin'; SHOW VARIABLES LIKE 'binlog_format'; SHOW VARIABLES LIKE 'binlog_row_image';

如果log_bin的值为OFF，需要修改配置并重启 MySQL。云数据库（如阿里云 RDS、腾讯云 CDB）通常会默认开启 binlog，但格式不一定满足 CDC 要求，需要根据控制台文档确认。

4.2 开启 binlog 并配置 ROW 格式

编辑 MySQL 配置文件/etc/my.cnf或自定义的 MySQL Docker 配置，在[mysqld]段添加以下内容：

[mysqld] server_id = 223344 log_bin = mysql-bin binlog_format = ROW binlog_row_image = FULL

如果使用 MySQL 5.7，可以设置 binlog 保留时间：

expire_logs_days = 14

如果使用 MySQL 8.0，推荐使用秒数配置：

binlog_expire_logs_seconds = 1209600

这里要注意：server_id必须全局唯一。如果多台机器连接同一个 MySQL 实例，或者多个 CDC 任务连接到同一个实例，它们的server_id不能重复，否则会导致连接断开。

修改配置后重启 MySQL：

systemctl restart mysqld

4.3 创建 CDC 专用账号

为了安全，不要用 root 账号直接采集 binlog。建议创建专用账号，只授予最小权限。

CREATE USER 'cdc_user'@'%' IDENTIFIED BY 'cdc_pass'; GRANT SELECT, RELOAD, SHOW DATABASES, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'cdc_user'@'%'; FLUSH PRIVILEGES;

这里每个权限都有自己的用途：

• SELECT：Debezium 做初始快照时，需要读取表数据。
• REPLICATION SLAVE：允许伪装成从库，请求 binlog。
• REPLICATION CLIENT：允许查询主库状态，例如SHOW MASTER STATUS。
• RELOAD：某些快照场景需要执行FLUSH TABLES WITH READ LOCK保证一致性。

4.4 验证 binlog 是否产生

重新登录 MySQL，执行：

SHOW MASTER STATUS;

正常情况下会看到类似下面的输出，说明 binlog 已开启：

+------------------+----------+--------------+------------------+-------------------+ | File | Position | Binlog_Do_DB | Binlog_Ignore_DB | Executed_Gtid_Set | +------------------+----------+--------------+------------------+-------------------+ | mysql-bin.000003 | 156 | | | | +------------------+----------+--------------+------------------+-------------------+

执行几条 DML 语句后，再用SHOW BINLOG EVENTS IN 'mysql-bin.000003' LIMIT 5;查看，就能看到 binlog 事件了。

5. 实战：Debezium 采集 MySQL 变更到 Kafka

5.1 用 Docker Compose 搭一套最小环境

为了方便复现，这里直接用 Docker Compose 启动 MySQL、Zookeeper、Kafka、Debezium Connect 四个组件。

创建docker-compose.yml：

version: "3.8" services: mysql: image: mysql:8.0 container_name: mysql environment: MYSQL_ROOT_PASSWORD: root MYSQL_DATABASE: shop command: - "--server-id=223344" - "--log-bin=mysql-bin" - "--binlog-format=ROW" - "--binlog-row-image=FULL" - "--binlog-expire-logs-seconds=1209600" ports: - "3306:3306" volumes: - mysql-data:/var/lib/mysql zookeeper: image: confluentinc/cp-zookeeper:latest container_name: zookeeper environment: ZOOKEEPER_CLIENT_PORT: 2181 ZOOKEEPER_TICK_TIME: 2000 kafka: image: confluentinc/cp-kafka:latest container_name: kafka depends_on: - zookeeper environment: KAFKA_BROKER_ID: 1 KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181 KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092 KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1 ports: - "9092:9092" connect: image: debezium/connect:latest container_name: connect depends_on: - kafka environment: BOOTSTRAP_SERVERS: kafka:9092 GROUP_ID: 1 CONFIG_STORAGE_TOPIC: connect-configs OFFSET_STORAGE_TOPIC: connect-offsets STATUS_STORAGE_TOPIC: connect-status KEY_CONVERTER: org.apache.kafka.connect.storage.StringConverter VALUE_CONVERTER: org.apache.kafka.connect.json.JsonConverter VALUE_CONVERTER_SCHEMAS_ENABLE: "false" ports: - "8083:8083" volumes: mysql-data:

启动服务：

docker-compose up -d

生产环境强烈建议锁定镜像 tag，而不是使用latest。这里使用latest只是为了演示方便。

5.2 准备测试表和数据

进入 MySQL 容器：

docker exec -it mysql mysql -uroot -proot shop

创建订单表：

CREATE TABLE orders ( id BIGINT PRIMARY KEY AUTO_INCREMENT, order_no VARCHAR(32) NOT NULL, user_id BIGINT NOT NULL, amount DECIMAL(10, 2) NOT NULL, status VARCHAR(20) NOT NULL, create_time DATETIME NOT NULL, update_time DATETIME NOT NULL ); INSERT INTO orders (order_no, user_id, amount, status, create_time, update_time) VALUES ('A001', 1001, 99.50, 'CREATED', NOW(), NOW());

5.3 注册 Debezium Connector

通过 Kafka Connect 的 REST API 注册 MySQL Connector：

curl -i -X POST \ -H "Accept:application/json" \ -H "Content-Type:application/json" \ http://localhost:8083/connectors/ \ -d '{ "name": "mysql-orders-connector", "config": { "connector.class": "io.debezium.connector.mysql.MySqlConnector", "database.hostname": "mysql", "database.port": "3306", "database.user": "cdc_user", "database.password": "cdc_pass", "database.server.id": "223345", "database.server.name": "mysql-demo", "database.include.list": "shop", "table.include.list": "shop.orders", "schema.history.internal.kafka.topic": "schema-changes.mysql", "snapshot.mode": "initial" } }'

这里有几个配置需要重点说明：

• database.server.id：Debezium 连接 MySQL 时使用的 server-id，必须与源库的server_id不同，并且与其他连接器保持唯一。
• database.server.name：决定 Kafka topic 的前缀。默认情况下，topic 名称为{server.name}.{database}.{table}。
• schema.history.internal.kafka.topic：Debezium 会把历史的 schema 信息存储在 Kafka 里，这样后续解析老 binlog 时，还能知道当时的表结构。如果使用 Debezium 1.x，配置项名称是database.history.kafka.topic。

等待几十秒后，检查连接器状态：

curl http://localhost:8083/connectors/mysql-orders-connector/status

看到"state": "RUNNING"说明连接器已正常运行。

5.4 观察 insert / update / delete 消息

Debezium 启动时会先执行一次一致性快照，已有数据会以op: "r"（read）的消息写入 Kafka。随后源库的增删改操作会实时产生消息。

先在 MySQL 中插入一条新数据：

INSERT INTO orders (order_no, user_id, amount, status, create_time, update_time) VALUES ('A002', 1002, 29.90, 'CREATED', NOW(), NOW());

启动 Kafka 自带的消费者查看消息：

docker exec -it kafka kafka-console-consumer.sh \ --bootstrap-server localhost:9092 \ --topic mysql-demo.shop.orders \ --from-beginning \ --max-messages 5

插入操作对应的消息大致如下（这里简化了 schema 部分）：

{ "before": null, "after": { "id": 2, "order_no": "A002", "user_id": 1002, "amount": 29.90, "status": "CREATED", "create_time": 1690000000000, "update_time": 1690000000000 }, "source": { "db": "shop", "table": "orders", "ts_ms": 1690000001000 }, "op": "c" }

再执行一次更新：

UPDATE orders SET status = 'PAID', update_time = NOW() WHERE order_no = 'A002';

更新操作对应的消息op变为u，同时before和after都会存在：

{ "before": { "id": 2, "order_no": "A002", "user_id": 1002, "amount": 29.90, "status": "CREATED", "create_time": 1690000000000, "update_time": 1690000000000 }, "after": { "id": 2, "order_no": "A002", "user_id": 1002, "amount": 29.90, "status": "PAID", "create_time": 1690000000000, "update_time": 1690000001000 }, "source": { "db": "shop", "table": "orders", "ts_ms": 1690000002000 }, "op": "u" }

关键是删除操作。执行：

DELETE FROM orders WHERE order_no = 'A002';

即使目标库没有做任何特殊处理，删除消息也会被 binlog 完整记录：

{ "before": { "id": 2, "order_no": "A002", "user_id": 1002, "amount": 29.90, "status": "PAID", "create_time": 1690000000000, "update_time": 1690000001000 }, "after": null, "source": { "db": "shop", "table": "orders", "ts_ms": 1690000003000 }, "op": "d" }

这就是 binlog 方案与周期同步最大的区别：删除操作不再丢失，更新也不依赖业务方是否维护update_time字段，一切以数据库真实发生的变更为准。

5.5 消息里每个字段的含义

Debezium 的消息体结构是固定的，理解这些字段对下游处理非常有帮助：

• before：数据变更前的镜像。插入操作中为null，更新和删除操作中可能存在。
• after：数据变更后的镜像。删除操作中为null。
• op：操作类型。c表示插入，u表示更新，d表示删除，r表示初始快照阶段读取。
• source.ts_ms：变更事件发生的时间戳（毫秒），通常对应 binlog 中记录的时间。
• source.db/source.table：变更事件所在的数据库和表。

下游拿到这些消息后，可以在目标侧执行 upsert、逻辑删除、物理删除等不同的数据策略，灵活度比周期同步高很多。

6. 从 Kafka 接入 BigQuery 的三条路线

把 MySQL 的变更数据送到 Kafka 只是第一步，接下来还需要将数据落地到 BigQuery。取决于团队的基础设施，有三条比较常用的路线。

6.1 路线一：Kafka Connect BigQuery Sink

社区中有开源的 Kafka Connect BigQuery Sink 连接器，常见的实现来自 GitHub 上的kafka-connect-bigquery项目。它的工作方式是在 Kafka Connect 集群中再部署一个 Sink 连接器，消费mysql-demo.shop.orders主题，把 JSON 消息写入 BigQuery。

注册 BigQuery Sink 的配置大致如下（不同版本参数会有差异，请以项目 README 为准）：

{ "name": "bigquery-sink", "config": { "connector.class": "com.wepay.kafka.connect.bigquery.BigQuerySinkConnector", "topics": "mysql-demo.shop.orders", "projectId": "your-gcp-project", "datasets": "orders=shop_ods", "keyfile": "/etc/bigquery/keyfile.json", "autoCreateTables": "true", "sanitizeTopics": "true" } }

这种方案适合已经使用 Kafka Connect 的团队，复用现有的连接器机制，运维成本相对低。需要注意，该连接器默认会使用 BigQuery 的流式插入接口，对数据量很大的场景需要考虑配额和费用。

6.2 路线二：Kafka → Pub/Sub → Dataflow → BigQuery

这是 GCP 原生生态里很常见的管道设计：Kafka 中的数据通过 Pub/Sub 连接器转发到 Pub/Sub 主题，然后由 Dataflow 作业消费 Pub/Sub 消息，经过必要的清洗和转换后写入 BigQuery。

这条链路看起来多了几个组件，但优势在于，一旦数据进入 Pub/Sub，后续就有了 GCP 托管的流处理能力。Dataflow 自带的窗口、去重、流式写入 BigQuery 语义都比较成熟，适合对数据延迟和准确性要求较高的生产链路。

6.3 路线三：Flink CDC 直连 BigQuery Connector

如果不想引入 Kafka 中间层，也可以直接用 Flink CDC 从 MySQL 采集数据，再通过 Flink 的 BigQuery Sink 写入。

Flink CDC 的源表定义非常简洁，下面是一个可运行的 Flink SQL 示例：

CREATE TABLE orders_cdc ( id BIGINT, order_no STRING, user_id BIGINT, amount DECIMAL(10, 2), status STRING, create_time TIMESTAMP(3), update_time TIMESTAMP(3), PRIMARY KEY (id) NOT ENFORCED ) WITH ( 'connector' = 'mysql-cdc', 'hostname' = 'localhost', 'port' = '3306', 'username' = 'cdc_user', 'password' = 'cdc_pass', 'database-name' = 'shop', 'table-name' = 'orders', 'server-id' = '5400' );

之后通过INSERT INTO语句将数据写入 BigQuery 目标表。Flink CDC 底层同样依赖 Debezium 解析 binlog，因此它具备 Debezium 的实时性和变更捕获能力，同时把数据处理逻辑放进同一条流中。

6.4 BigQuery 侧的前置准备

无论使用哪条路线，访问 BigQuery 前都需要完成以下准备工作：

1. 在 GCP 中创建项目，并启用 BigQuery API。
2. 创建服务账号，授予 BigQuery Data Editor 或更细粒度的权限。
3. 下载服务账号的 JSON Key，部署到 Kafka Connect 或 Dataflow 所在节点。
4. 在 BigQuery 中提前创建数据集和表。

BigQuery 侧的目标表可以直接参考 MySQLorders表的结构来建。如果对字段类型不敏感，也可以让连接器自动建表，但生产环境更推荐由 DBA 和数据工程师统一设计，明确分区字段和 clustering 字段。

7. 周期性同步 vs CDC：一个对比表

为了更直观地说明两种方案差异，这里做一个汇总：

表格里没有绝对的“谁更好”，因为周期性同步在小数据量、非核心场景下仍有成本优势。但当数据量和实时性要求上来之后，binlog 方案的优势会越来越明显。

8. 常见问题与排查思路

8.1 常见错误对照表

以下是 CDC 链路搭建过程中最常遇到的几类问题：

8.2 三个高频场景排查

场景一：快照完成但增量不生效。

检查连接器的snapshot.mode。默认initial表示先做快照再读增量。如果表数据量很大，快照阶段会比较长，这期间的新增变更不会丢失，Debezium 会在快照完成后继续从记录的位置读取 binlog。如果设置了schema_only，则不会复制已有数据，只有后续增量。

场景二：topic 里出现大量d操作，目标库不想物理删除。

删除消息在目标侧可以有两种处理方式：直接删除目标行，或者做逻辑删除（例如在目标表增加is_deleted字段并标记）。具体策略需要和目标表的下游消费方确认。

场景三：binlog 消费断点过期。

假设 CDC 任务停了三天，而 MySQL 的 binlog 只保留两天，重新启动后会出现“binlog 文件找不到”的报错。这种情况只能重新做一次全量快照。因此生产环境的 binlog 保留时间必须覆盖“故障恢复时间 + 下游排查时间”，建议配置 7 天以上，并根据数据量评估磁盘成本。

9. 落地建议与下一步

把周期性同步换成 CDC，不是技术炫技，而是当数据体量和实时性要求上来之后，早晚要迈过去的一步。建议先从一张非核心订单表开始实验，按照本文第四、五章的步骤搭建最小链路，验证 insert、update、delete 三类消息都能正常到达下游，再逐步推广到核心表。

具体落地时，有几点可以优先确认：

1. 先保证 binlog 保留时间合规。这是最容易在接入后期踩坑的问题，建议上线前就和 DBA 确认清楚。
2. 账号权限遵循最小化原则。CDC 账号只给 SELECT 和复制相关权限，不要使用 root。
3. 监控要跟上。重点监控 Kafka 消费延迟、BigQuery 写入失败率、MySQL 连接数变化。
4. 定期做故障演练。停掉 Kafka 一段时间再恢复，观察 CDC 任务能否从 binlog 断点继续消费，这比任何文档都更有说服力。

下一步可以继续深入了解 Debezium 的 schema 变更处理机制、Flink CDC 的 checkpoint 语义，以及 BigQuery 流式写入的成本控制。数据同步链路只要跑起来，就会出现新的问题，把每个问题记录成排查笔记，是工程师成长最快的方式。