企业官网建设流程全解析

FlinkCDC这套东西，说实话，我最早接触的时候也以为只是又一个数据同步工具，结果真正上手搞了一次从 MySQL 到 Elasticsearch 的实战之后，才意识到它解决的不仅仅是“把数据搬过去”这么简单。日常业务里，MySQL 扛住了事务和一致性，但碰上模糊搜索、聚合分析、大范围检索这些场景就明显吃力，这时候把数据同步到 ES 里做查询加速，几乎是标配玩法。而 FlinkCDC 最大的价值，就是让你不用写一堆定时任务，也不用自己去解析 binlog，直接用现成的框架把变更事件流出来，再交给 Flink 做加工、过滤、转换，最后落到 ES。

这篇文章就是一次完整的实操记录，从环境准备、CDC 原理、代码实现到线上坑点排查，全部基于我自己跑通的案例。不管你是刚接触 CDC 的新手，还是已经用过一段时间但被各种诡异问题折腾过的人，这篇文章都能给你一些可以直接抄的答案。

1. 整体设计思路：为什么选 FlinkCDC 而不是其他方案

1.1 同步方案的选型对比

做 MySQL 到 ES 的同步，市面上能选的方案其实不少，我简单列一下我实际比较过的几条路线：

• 直接写业务双写：在应用层每次更新 MySQL 的同时更新 ES。这个方案最大的问题就是耦合太重，业务代码里到处塞同步逻辑，一旦 ES 出问题还会拖垮主流程。
• 用定时任务扫表同步：实现简单，但延迟高，而且增量数据很难精准捕捉，删除操作更难处理。
• 用 Canal 监听 binlog 再转投 ES：Canal 本身很成熟，但你需要额外维护一套中间件，还要自己写消费者逻辑，并且 Canal 的部署和配置有一定门槛。
• 用 FlinkCDC 直接入 ES：FlinkCDC 可以直连 MySQL 的 binlog，把变更事件转成 DataStream，再用 Flink 的算子做处理，最后通过连接器写入 ES。

我最终选了 FlinkCDC，核心原因有三个：

第一，实时性足够高。binlog 是 MySQL 的物理变更日志，只要事务提交，FlinkCDC 立刻就能感知到，秒级甚至毫秒级延迟，这比任何定时扫表方案都强。

第二，处理能力灵活。Flink 本身提供了丰富的算子，你可以对 CDC 流出来的数据做任意加工，过滤字段、清洗格式、关联维表、聚合统计，全都能在一个 Pipeline 里完成。

第三，状态管理和容错机制完善。Flink 的 checkpoint 机制可以记录消费位点，任务重启后能从上次的位置继续读，不会丢数据也不会重复读太多。

1.2 FlinkCDC 的核心原理简述

很多人用 FlinkCDC 只是当成黑盒用，但我觉得至少要把原理捋一遍，否则出了问题根本不知道从哪里排查。

FlinkCDC 底层实际上是封装了 Debezium，它会伪装成一个 MySQL 从节点，向主节点发送 dump 请求，然后持续接收 binlog 事件。每一条 binlog 事件都会被解析成结构化的变更记录，包含 before、after、op 等关键信息。

这里有个容易混淆的点：FlinkCDC 有两种模式，一种是有状态的全量加增量模式，另一种是纯增量模式。默认情况下，FlinkCDC 启动时会先执行一次全量 snapshot，把当前表已有的数据全部读出来，然后再无缝切换到 binlog 增量监听。这个过程对用户是透明的，你只需要指定一个 offset 或者让它自动管理。

从表结构的角度看，FlinkCDC 输出的数据格式大致是这样的：

{ "before": null, "after": { "id": 1001, "name": "张三", "age": 28 }, "op": "c" }

op 字段标识操作类型，c 是新增，u 是更新，d 是删除，r 是快照读取。写 ES 的时候，你通常会把 before 和 after 合并处理，根据 op 决定是索引、更新还是删除。

1.3 为什么这套方案适合大多数人

如果你只是同步几张表到 ES 用于搜索，规模不大，并发不高，那 FlinkCDC 绝对是性价比最高的选择。它的部署成本比 Canal 低，不需要额外引入消息队列，写代码的量也比自己解析 binlog 少得多。

而且 Flink 生态的连接器非常丰富，MySQL 到 ES 只是其中一种，完全可以一套代码跑多个同步任务，后续扩展其他数据源或者目标存储也都方便。

2. 环境准备与核心依赖配置

2.1 版本选型这里最容易踩坑

版本问题是我见过最多的坑。FlinkCDC 的版本、Flink 的版本、连接器的版本，三者必须匹配，否则编译都过不去，或者运行时报一堆类找不到。

我这次使用的组合是：

• Flink：1.17.1
• FlinkCDC：2.3.0
• MySQL：5.7.35
• Elasticsearch：7.15.2
• Java：1.8

先说为什么这么选。FlinkCDC 2.3.0 官方支持 Flink 1.13 到 1.17 的多个版本，但我实测下来 1.17 最稳。ES 连接器方面，我用的是 flink-connector-elasticsearch7，注意这个连接器的版本要和 Flink 版本对应，否则会出现冲突。

MySQL 5.7 和 8.0 在 binlog 格式上有些差异，如果你用的是 MySQL 8.0，需要额外确认 binlog_row_image 的参数设置，后面我会专门提到。

2.2 MySQL 必须开启的参数

FlinkCDC 读取 binlog 依赖几个 MySQL 参数，没开的话任务会直接报错或者什么都读不到。

首先是 binlog 开关，MySQL 默认可能是关闭的，需要确认配置：

server-id=1 log_bin=mysql-bin binlog_format=ROW binlog_row_image=FULL expire_logs_days=7

binlog_format 必须设成 ROW，因为只有行级别的 binlog 才能记录每一行数据的前后变化。binlog_row_image 设成 FULL 也很关键，如果设成 MINIMAL，那 binlog 里可能只记录被修改的列，其他列的值是空的，同步到 ES 后数据就不全了。

注意：修改这些参数后必须重启 MySQL 才会生效。另外建议给 FlinkCDC 单独建一个账号，授予 SELECT、REPLICATION SLAVE、REPLICATION CLIENT 三个权限，不要用 root。

然后就是授权：

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

2.3 引入依赖的完整清单

我用的是 Maven 工程，依赖如下：

<dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-java</artifactId> <version>1.17.1</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-streaming-java</artifactId> <version>1.17.1</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-clients</artifactId> <version>1.17.1</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-connector-elasticsearch7</artifactId> <version>3.0.1-1.16</version> </dependency> <dependency> <groupId>com.ververica</groupId> <artifactId>flink-connector-mysql-cdc</artifactId> <version>2.3.0</version> </dependency>

注意 flink-connector-elasticsearch7 的版本后缀是 -1.16 或 -1.17，这个要跟 Flink 主版本匹配。如果你发现运行时出现 Jackson 相关的类冲突，多半是依赖版本没对齐，建议把 flink-connector-elasticsearch7 里传递进来的 Jackson 依赖排除掉，用 Flink 自带的。

3. 核心代码实现：从 MySQL 到 ES 的完整链路

3.1 构建 FlinkCDC 数据源

代码部分我直接给出一个可以跑的 Demo。第一步是从 MySQL 读取变更事件：

import com.ververica.cdc.connectors.mysql.source.MySqlSource; import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema; import org.apache.flink.api.common.eventtime.WatermarkStrategy; import org.apache.flink.streaming.api.datastream.DataStreamSource; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; public class MysqlToEsJob { public static void main(String[] args) throws Exception { StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.enableCheckpointing(5000L); env.setParallelism(1); MySqlSource<String> mySqlSource = MySqlSource.<String>builder() .hostname("127.0.0.1") .port(3306) .databaseList("test_db") .tableList("test_db.user") .username("cdc_user") .password("your_password") .deserializer(new JsonDebeziumDeserializationSchema()) .startupOptions(StartupOptions.initial()) .build(); DataStreamSource<String> streamSource = env.fromSource( mySqlSource, WatermarkStrategy.noWatermarks(), "mysql-cdc-source" ); streamSource.print(); env.execute("mysql-cdc-to-es"); } }

这段代码里有几个关键点：

• databaseList 和 tableList 可以用正则匹配，如果不写 tableList，默认监听 databaseList 里所有表。
• startupOptions.initial() 表示先做全量再增量，如果只想监听增量，可以用 latest()。
• JsonDebeziumDeserializationSchema 会把 CDC 事件直接序列化成 JSON 字符串，适合快速开发。如果后续要做复杂的字段映射，我建议自定义反序列化器，后面会讲。

先用 print() 跑一下，你能在控制台看到类似这样的输出：

{ "before": null, "after": { "id": 1, "name": "小明", "age": 25, "city": "上海" }, "op": "c" }

看到这个输出就说明 CDC 链路已经通了。

3.2 自定义反序列化器，按需清洗数据

JSON 字符串虽然方便，但直接写到 ES 会有几个问题：

• 字段名可能不符合 ES 的索引规范。
• 嵌套结构无法优雅处理。
• before 和 after 需要自己合并，比较繁琐。

所以我更推荐实现一个自定义的 DeserializationSchema，直接把数据转成自己定义的实体类或者 JSONObject，同时把 op 字段剥离出来。

这里我给出一个更实用的做法，转成 JSONObject 并且只保留 after 中的数据：

import com.alibaba.fastjson2.JSONObject; import com.ververica.cdc.debezium.DebeziumDeserializationSchema; import org.apache.flink.api.common.typeinfo.TypeInformation; import org.apache.flink.util.Collector; import org.apache.kafka.connect.source.SourceRecord; import org.apache.kafka.connect.json.JsonConverter; import org.apache.kafka.connect.storage.ConverterType; import java.util.HashMap; import java.util.Map; public class CustomDeserializationSchema implements DebeziumDeserializationSchema<JSONObject> { private static final JsonConverter CONVERTER = new JsonConverter(); @Override public void deserialize(SourceRecord record, Collector<JSONObject> out) throws Exception { Map<String, Object> config = new HashMap<>(); config.put(ConverterType.KEY_CONFIG, "org.apache.kafka.connect.json.JsonConverter"); CONVERTER.configure(config, false); byte[] bytes = CONVERTER.fromConnectData(record.topic(), record.valueSchema(), record.value()); JSONObject jsonObject = JSONObject.parseObject(new String(bytes)); JSONObject after = jsonObject.getJSONObject("after"); String op = jsonObject.getString("op"); if ("d".equals(op)) { JSONObject before = jsonObject.getJSONObject("before"); out.collect(before); } else { out.collect(after); } } @Override public TypeInformation<JSONObject> getProducedType() { return TypeInformation.of(JSONObject.class); } }

这里有个细节，删除事件里 after 是 null，只有 before 有值，所以对于删除操作要特殊处理，否则你得到的就是一个空对象。

提示：如果你不需要处理删除事件，可以在同步链路里直接过滤掉 op=d 的数据。但大部分实战场景里，MySQL 删除了数据，ES 里对应的文档也应该删除，否则就会出现两边数据不一致。

3.3 写 ES 连接器的关键配置

拿到清洗后的数据之后，下一步就是写入 Elasticsearch。核心代码如下：

import org.apache.flink.streaming.connectors.elasticsearch7.ElasticsearchSink; import org.apache.flink.streaming.connectors.elasticsearch7.RestClientFactory; import org.apache.http.HttpHost; import org.elasticsearch.action.delete.DeleteRequest; import org.elasticsearch.action.index.IndexRequest; import org.elasticsearch.client.Requests; import java.util.HashMap; import java.util.Map; public class EsSinkBuilder { public static ElasticsearchSink<JSONObject> buildEsSink() { Map<String, String> config = new HashMap<>(); config.put("bulk.flush.max.actions", "1000"); config.put("bulk.flush.max.size.mb", "5"); config.put("bulk.flush.interval.ms", "2000"); return new ElasticsearchSink<>( config, new HttpHost("127.0.0.1", 9200, "http"), new RestClientFactory() { @Override public void configureRestClientBuilder(RestClientBuilder builder) { // 如果需要身份认证，在这里添加 } }, new ElasticsearchSinkFunction<JSONObject>() { @Override public void process(JSONObject element, RuntimeContext ctx, RequestIndexer indexer) { String op = element.getString("_op"); String id = element.getString("id"); element.remove("_op"); if ("d".equals(op)) { DeleteRequest deleteRequest = new DeleteRequest("user_index", id); indexer.add(deleteRequest); } else { IndexRequest indexRequest = new IndexRequest("user_index") .id(id) .source(element.toJsonString(), XContentType.JSON); indexer.add(indexRequest); } } } ); } }

关于这个 sink，我特别想强调几个参数：

• bulk.flush.max.actions：攒够 1000 条后批量写入。如果单条数据很大，可以调小一点，比如 500。
• bulk.flush.max.size.mb：批量数据体积上限，默认 5MB 够了。
• bulk.flush.interval.ms：即使没攒够条数，每隔 2 秒也会强制刷一次，避免长时间没有数据落盘。

这三个参数共同决定了写入吞吐和延迟的平衡。如果你追求极低的延迟，可以把 interval.ms 调成 500，副作用是请求数量变多，ES 压力会大一点。如果追求高吞吐，可以把 actions 调到 5000，前提是你的 ES 集群扛得住。

3.4 主程序串联完整链路

把数据源和 sink 串起来，完整的主程序就是：

public class MysqlCdcToEsJob { public static void main(String[] args) throws Exception { StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.enableCheckpointing(10000L); MySqlSource<JSONObject> source = MySqlSource.<JSONObject>builder() .hostname("127.0.0.1") .port(3306) .databaseList("test_db") .tableList("test_db.user") .username("cdc_user") .password("your_password") .deserializer(new CustomDeserializationSchema()) .startupOptions(StartupOptions.initial()) .build(); DataStreamSource<JSONObject> stream = env.fromSource( source, WatermarkStrategy.noWatermarks(), "mysql-cdc-source" ); stream.sinkTo(EsSinkBuilder.buildEsSink()); env.execute("mysql-cdc-to-es-job"); } }

这里有个容易忽略的点：env.setParallelism(1) 需要根据你的数据量来定。如果你设置并行度为 2 或更高，FlinkCDC 会起多个分片读取 binlog，这时 MySQL 端需要配置足够的 server-id 去区分不同的连接。我建议初期就用并行度 1 跑通，后面再根据压力测试结果调并行度。

4. 实操过程中的参数优化与模式选择

4.1 全量加增量模式的实际表现

我这次用的是 StartupOptions.initial()，启动后的表现是先花了几秒扫描表数据，然后立刻进入增量监听状态。如果你表的数据量很大，比如千万级，全量阶段可能会持续几分钟甚至更久，这段时间 binlog 里的新增数据会不会丢？

答案是不会。FlinkCDC 的全量阶段会先记录一个 binlog 位点，然后扫描全量数据，扫描完成后再从那个位点接着读增量数据，所以不会丢数据。但这也有个隐患：全量扫表期间，如果源表的数据被并发修改得很频繁，那同步时间会变长，因为 FlinkCDC 需要处理的数据量变大了。

另外一个细节是，全量阶段建议给源表加一个稳定的排序字段，因为 FlinkCDC 的分片算法默认会基于主键或者唯一键拆分 chunk。没有主键的表，FlinkCDC 处理起来会慢很多，甚至直接不支持。所以源表一定要有主键，这是硬性要求。

4.2 只同步增量数据怎么做

如果你的业务场景是：ES 里已经有一份初始化好的全量数据，只需要从某个时间点开始同步增量，那你就用 StartupOptions.latest()。这个模式启动后不会扫描历史数据，直接接 binlog 当前及以后的事件。

这个模式的好处是启动速度快，但危险在于：任务启动之前，MySQL 里如果已经发生了变更，这些变更你永远不会知道。所以除非你确定 ES 里的数据已经和 MySQL 完全一致，否则我仍然建议用 initial()。

还有一种场景是：任务因为故障停机了几天，你重启后希望从停机之前的位点继续读。这时候你就需要用到指定位点启动：

.startupOptions(StartupOptions.specificOffset("mysql-bin.000123", 456789L))

位点的来源就是 Flink 的 checkpoint 记录。如果你启用了 checkpoint，重启任务时会自动从最近一次成功的 checkpoint 恢复，不需要手动指定，所以务必不要关闭 checkpoint。

4.3 写入 ES 的索引设计建议

ES 索引的 mapping 设计也是同步质量的关键。如果你直接让 ES 自动推断字段类型，可能会出现数字被识别成 keyword、日期格式混乱等问题。

我调整后的 mapping 大概是这样的：

{ "mappings": { "properties": { "id": { "type": "long" }, "name": { "type": "text", "analyzer": "ik_max_word" }, "age": { "type": "integer" }, "city": { "type": "keyword" }, "create_time": { "type": "date", "format": "yyyy-MM-dd HH:mm:ss" } } } }

这里面有几个要点：

• 需要被模糊搜索的字段，比如 name，用 text 类型，并配置中文分词器。
• 需要精确匹配或者聚合的字段，比如 city，用 keyword。
• 日期字段必须明确指定 format，否则 ES 默认只接受标准 ISO 格式。
• id 字段建议用 long，避免 ES 自动映射成 text 导致排序和过滤异常。

提示：ES 索引的 mapping 创建之后就不能随意修改字段类型，所以建索引之前一定要先想清楚你的查询需求。

5. 线上排查实录：常见问题与解决思路

5.1 连接 MySQL 时报 Access denied

这个问题多数是账号权限不对，或者 FlinkCDC 请求时用的 server-id 和别的客户端冲突。

排查步骤：

1. 先手动用账号连一下 MySQL，确认密码和 host 没问题。
2. 执行 SHOW GRANTS FOR 'cdc_user'@'%'; 确认有 SELECT、REPLICATION SLAVE、REPLICATION CLIENT。
3. 如果多个 FlinkCDC 任务共用同一个 MySQL，建议每个任务显式指定不同的 server-id，避免互相踢掉连接。

FlinkCDC 的 server-id 可以这样配置：

.serverId("5400-5405")

5.2 全量阶段完成后增量数据一直不更新

这个坑浪费了我不少时间，最后发现是 MySQL 的 binlog 清理策略导致的。如果 expire_logs_days 设置太短，全量扫描期间的历史 binlog 已经被清理，FlinkCDC 就无法从记录的位点继续读取增量数据。

解决方案：

• 增大 expire_logs_days，比如 7 天以上。
• 全量数据量特别大的时候，尽量在业务低峰期执行首次同步，降低 binlog 被覆盖的风险。
• 也可以用 binlog 文件大小来控制，比如 max_binlog_size=512M，确保每个文件不会太大。

5.3 ES 写入报 document parsing exception

这个问题的典型场景是：MySQL 中某个字段的值是 null，或者字段类型和 ES mapping 不匹配。

比如 MySQL 里 create_time 是字符串，但写入 ES 时格式不是 yyyy-MM-dd HH:mm:ss，就会导致解析失败。

解决办法是在反序列化器里做一层数据清洗，把所有字段都转成合适的类型，或者把类型不匹配的字段强制转成字符串：

if (after.get("create_time") == null) { after.put("create_time", ""); } else { after.put("create_time", after.get("create_time").toString()); }

ES 对脏数据的容忍度其实很低，一个字段解析失败就会导致整条文档写入失败，所以清洗这层不能省。

5.4 序列化：实体类报反序列化失败

如果你把反序列化器配成指定实体类，比如 UserBean，那么 MySQL 表新增字段后，这个实体类如果没有同步增加字段，反序列化同样会出问题。

我个人的建议是尽量用 Map 或 JSONObject 作为中间格式，不做强类型映射。虽然类型安全差点，但胜在灵活，字段变更不用改代码重新上线。

5.5 数据重复写入怎么处理

Flink 的检查点恢复机制是 at-least-once 语义，所以极端情况下会出现重复数据。如果你对数据准确性要求极高，可以通过给每条数据加一个唯一 ID 或者使用 ES 的 doc id 来做幂等。

我这边因为用 id 作为 ES 文档的 _id，所以重复写入只会覆盖，不会出现两条重复文档，这个问题天然被规避掉了。

6. 超实用的几个技巧与升级思路

6.1 多表同步的简化处理

如果你要同时同步 user 表和 order 表到 ES 的两个索引，最直接的办法是写两个 source 和两个 sink。但更好的做法是用一个 source 监听多个表，然后在 process 函数里根据表名分发到不同的索引：

@Override public void process(JSONObject element, RuntimeContext ctx, RequestIndexer indexer) { String tableName = element.getString("table"); if ("user".equals(tableName)) { // 写入 user_index } else if ("order".equals(tableName)) { // 写入 order_index } }

这个方案减少了任务数，资源占用也更低。

6.2 同步过程中做字段过滤

有些字段比较敏感，比如用户手机号，你不想直接同步到 ES。你可以在反序列化步骤里直接 remove 掉：

after.remove("phone"); after.remove("password");

这样做的好处是数据从源头就被清理掉了，ES 里根本不可能出现这些字段，安全和合规上更有保障。

6.3 用 SQL 模式写 FlinkCDC

如果你不想写那么多 Java 代码，FlinkCDC 2.x 还支持 Flink SQL 方式，几条 SQL 就能完成同步：

CREATE TABLE user_source ( id INT PRIMARY KEY NOT ENFORCED, name STRING, age INT, city STRING ) WITH ( 'connector' = 'mysql-cdc', 'hostname' = '127.0.0.1', 'port' = '3306', 'username' = 'cdc_user', 'password' = 'your_password', 'database-name' = 'test_db', 'table-name' = 'user' ); CREATE TABLE user_sink ( id INT PRIMARY KEY NOT ENFORCED, name STRING, age INT, city STRING ) WITH ( 'connector' = 'elasticsearch-7', 'hosts' = 'http://127.0.0.1:9200', 'index' = 'user_index', 'sink.bulk-flush.max-actions' = '1000', 'format' = 'json' ); INSERT INTO user_sink SELECT * FROM user_source;

SQL 方式特别适合团队里不熟悉 Java 的同事，逻辑一目了然，运维成本也低。我自己在快速验证的时候也经常先用 SQL 方式跑通，再决定要不要转成 DataStream API。

6.4 关于状态后端的建议

如果同步的数据量很大，建议把 Flink 的状态后端从默认的 HashMapStateBackend 改成 RocksDBStateBackend：

env.setStateBackend(new RocksDBStateBackend("hdfs:///flink/checkpoints"));

RocksDB 把状态存在本地磁盘，内存压力小很多，适合大批量同步任务。如果没有 HDFS，也可以直接配一个本地路径，但生产环境还是建议用分布式存储，避免单点故障后无法恢复。

7. 一些我自己常用的配置速查表

经常被问到各种参数怎么配，我整理了一份表，直接抄就行了。

配过一遍之后，你会发现这套链路其实不难，但每一个细节都值得认真对待。

我个人的体会是，FlinkCDC 真正降低的是实时同步的工程门槛，但它没法替你解决所有数据质量问题。无论做多少技术选型和参数调优，最后还是要回归到对源表数据的充分理解上。MySQL 里每一张要同步的表，哪些字段要保留、哪些字段要清洗、主键是否稳定、更新频率高不高，这些业务层面的判断，才是决定一个同步任务能不能长期稳定跑下去的根本。先把其中一张核心表跑通，再逐步扩展，是我最推荐的上手路径。