企业官网建设流程全解析

1. 这条链路的真实面貌：谁在解决什么问题

第一次把SqlServerCDC 数据源接到 Elasticsearch的时候，我以为最难的是写 Flink 作业，真正做下来才发现，作业代码反而是整条链路里最省心的部分。麻烦集中在两头：源库那边的 CDC 有没有真的跑起来、变更表有没有被清理作业悄悄截断；ES 那边的索引 mapping 有没有提前定好、写进去的文档 _id 是不是稳定。中间那段 Flink，只要参数对齐、checkpoint 配好，其实跑起来很稳。

这篇内容讲的就是SQL Server 业务表 → Flink CDC 捕获变更 → Elasticsearch 检索这条同步链路，属于 Flink CDC 系列的第二篇，重点从“数据源”这个角度切入，把 SQL Server CDC 的机制、Flink Source 参数、数据整形、ES Sink 调优和排查经验串成一条完整的线。适合两类人看：一类是正在做异构数据同步、想用 CDC 替代定时跑批的开发者；另一类是把作业跑起来了但老是丢数据、重复写、同步延迟越来越大的同学。文中涉及的表名、库名、服务器地址全部是虚构示例，替换成你自己的环境即可。

我先把这条链路的骨架摆出来，后面每一章都往这个骨架里填肉。整条链路分三段：

• 捕获段：SQL Server 开启数据库级和表级 CDC，Agent 作业把事务日志扫进cdc.<schema>_<table>_CT变更表，Flink CDC Source 通过 Debezium 的 SQL Server 实现读取这些变更。
• 整形段：Flink 里做字段裁剪、类型转换、脏数据处理、主键拼接，把 CDC 事件（带 before/after 的变更记录）拍平成一条可以整体覆盖的文档。
• 落库段：ES Sink 按主键做整文档覆盖写，批量提交、失败退避重试，配合 checkpoint 达成“结果层面的一次性”。

1.1 一个典型场景长什么样

举个我参与过的例子。某电商团队的订单主表在 SQL Server 上，日均订单量在几十万条量级，表里有两百多万行存量数据。下游有一套基于 ES 的订单检索页，运营和客服要靠它按订单号、手机号后四位、下单时间区间、订单状态做组合查询，要求延迟控制在秒级。

最早的方案是每五分钟跑一次增量 SQL，用update_time > 上次时间捞数据。这个方案在小数据量下能用，数据涨上来之后问题全冒出来了：update_time没有索引的时候全表扫，加了索引又影响写入；更致命的是“物理删除”捞不到，订单被删了 ES 里还留着，客服查出来一堆幽灵订单；还有一种情况是同一行在一个窗口里被改了很多次，增量 SQL 只取到最后一次，中间的变更历史丢了，虽然对检索页影响不大，但对账的时候就很难受。

换成 CDC 之后，这三个问题一次性解决：删除事件会以op = d的形式发出来，可以做逻辑删除也可以物理删除；变更的粒度是行级、顺序的，同一行改多少次就来多少条事件；对源库的压力从“周期性扫表”变成了“读事务日志”，对业务库几乎无感。

1.2 这套组合的能力边界

这里需要先立个边界，免得大家做完才发现预期不对。

其中“DDL 跟踪弱”这一条值得单独说一句。SQL Server 的 CDC 在表结构变更时行为比较别扭，加列之后新的列通常要重建 capture instance 才能被捕获，直接改表可能出现“新列的值一直是 null”的情况。所以我在实际项目里的做法是：把表结构变更当成一次发布来做，先停作业、改表、重建 capture instance、再启动，用scan.startup.mode = initial或者手动补一段数据来补齐。听起来笨，但比半夜被叫起来查“为什么新字段没同步”要省心。

2. SQL Server 这一侧：CDC 不是开关，是账本

很多人以为开了 CDC 就完事了，其实 CDC 在 SQL Server 里更像是一本账：Agent 作业是记账员，变更表是账本，清理作业是销毁旧账的。这三者任何一个停摆，上游看起来一切正常，下游拿到的数据却是断的。

2.1 CDC 在库里究竟写了什么

开启数据库级 CDC 之后，SQL Server 会在库里建一个cdcschema，里面有一堆系统表：

• cdc.change_tables：记录每张被跟踪表的变更表信息，包括 capture instance 名称。
• cdc.lsn_time_mapping：LSN 和时间戳的映射，增量位点就是靠它换算的。
• cdc.captured_columns/cdc.index_columns：元数据。
• cdc.dbo_<表名>_CT：真正的变更表，每张被跟踪的表对应一张。

变更表里除了原表的全部列，还有五个元数据列，这些列是理解 CDC 的关键：

这里有个特别容易踩的点：__$operation = 3和4是成对出现的，一次 update 在变更表里是两行，一行是更新前的镜像，一行是更新后的镜像。Flink CDC 的 Debezium 实现会帮你把这两个合并成一条带 before/after 的事件，所以你在 Flink 里看到的是op = u一条，而不是两条。但如果自己直接查变更表做同步，一定要记得处理这个成对关系，不然会写出双倍数据。

还有一个更隐蔽的问题：如果表没有主键或者唯一索引，CDC 在某些场景下无法正确定位行，Debezium 侧会直接报错或者行为异常。所以做 CDC 之前先检查一遍：

SELECT t.name AS table_name FROM sys.tables t LEFT JOIN sys.indexes i ON t.object_id = i.object_id AND i.is_primary_key = 1 WHERE i.index_id IS NULL;

查出来有结果的表，要么补主键，要么在 connector 里显式配置heartbeat之类的兜底，但说实话，补主键是最干净的路。

2.2 开启命令与权限的坑

开启分成两步，必须按顺序来。

-- 第一步：数据库级 USE [orderdb]; EXEC sys.sp_cdc_enable_db; GO -- 验证 SELECT name, is_cdc_enabled FROM sys.databases WHERE name = 'orderdb';

第二步是表级，这一步的参数比想象中重要：

EXEC sys.sp_cdc_enable_table @source_schema = N'dbo', @source_name = N'order_info', @role_name = NULL, @capture_instance = N'dbo_order_info', @supports_net_changes = 1, @filegroup_name = N'CDC';

逐个说下这些参数为什么这么选。

@role_name = NULL是给谁访问变更表设置门槛。设成 NULL 表示不限制，任何有 schema 访问权的人都能查，这在开发环境方便，在生产环境偏松。更稳的做法是建一个专用的数据库角色，把@role_name设成这个角色，然后只给同步账号这个角色。我一开始图省事用了 NULL，后来安全审计提了一嘴，改成了专用角色，改动不大但心里踏实。

@capture_instance是 capture instance 的名字，默认就是<schema>_<table>。这个名字直接决定了 Debezium 侧读的是哪个 instance，如果你自定义了名字，Flink connector 那边不需要额外配置，但你自己用cdc.fn_cdc_get_all_changes_<capture_instance>手工查的时候必须用对名字。我建议不要自定义，保持默认，减少一个出错点。

@supports_net_changes = 1会额外建一个 net changes 查询函数，Debezium 在某些增量阶段会用到。多占一点存储，但换来的是兼容性，不省这个。

@filegroup_name如果指定了，变更表会建在独立的文件组上。这个在我看来是必做项：变更表的写入模式和业务表完全不同（高频小事务追加），放在同一个文件组上会跟业务表的 IO 互相干扰。独立文件组 + 独立磁盘，是压力上来之后最有效的隔离手段。

权限这块有个很实在的坑：sp_cdc_enable_table要求调用者至少是db_owner或者sysadmin。很多团队给同步账号的是db_datareader+db_datawriter，执行这条就会报权限不足。我的处理方式是：开启动作由 DBA 执行一次，同步账号只给读权限，日常作业用读账号跑，这样职责清晰，也避免了同步账号权限过大的问题。

最后别忘了确认 Agent 在跑：

EXEC sys.sp_cdc_help_jobs;

输出里应该能看到两个作业：cdc.orderdb_capture和cdc.orderdb_cleanup。前者是采集，后者是清理。如果 Agent 服务停了，采集作业不会执行，变更表一直是空的，而数据库层面看起来“CDC 已启用”，极具迷惑性。我在一个测试环境上就因为这个排查了快一个小时。

2.3 三个决定成败的库级设置

除了开 CDC，还有几个库级设置会直接影响这条链路能不能跑通、跑得稳不稳。

第一，恢复模式。CDC 依赖事务日志，如果数据库是 SIMPLE 恢复模式，日志会被频繁截断，CDC 采集作业来不及扫的时候就可能丢 LSN。生产环境用 FULL 模式，配合定期日志备份，这是标准做法。

第二，快照隔离。Flink CDC 在做全量阶段的时候会对源表做一致性读，如果源库上还在跑 OLTP 业务，会遇到锁等待。开启快照隔离可以让读操作走行版本，不阻塞写入：

ALTER DATABASE [orderdb] SET ALLOW_SNAPSHOT_ISOLATION ON; ALTER DATABASE [orderdb] SET READ_COMMITTED_SNAPSHOT ON;

第二条READ_COMMITTED_SNAPSHOT打开之后，默认的读已提交隔离级别就会走行版本，效果立竿见影。要注意的是打开这个选项需要短暂独占，最好在维护窗口做。我在一个数据量比较大的库上开这个，等了大概几十秒，期间业务写入是阻塞的，所以别在高峰期点。

第三，清理作业的保留窗口。默认保留 4320 分钟（3 天）。这个值决定了“如果 Flink 作业挂了两天，回来还能不能续上”。如果你的作业可能因为发布、故障停超过一天，把这个值调到 10080（7 天）更稳妥：

EXEC sys.sp_cdc_change_job @job_type = N'cleanup', @retention = 10080;

代价是变更表占用的空间变大。要定期看占用：

SELECT OBJECT_NAME(object_id) AS table_name, SUM(reserved_page_count) * 8 / 1024 AS size_mb FROM sys.dm_db_partition_stats WHERE object_id IN ( SELECT object_id FROM sys.tables WHERE schema_id = SCHEMA_ID('cdc') ) GROUP BY object_id;

这里给个真实教训：有个环境变更表涨到了三十多 G，原因不是保留窗口设得长，而是 Flink 作业的位点停在一个很老的位置，Debezium 每次启动都从那个 LSN 往回补，补完还没提交就重启，循环往复。所以变更表体积异常增大，第一反应应该是去查同步作业的位点，而不是去砍保留窗口。

3. Flink CDC Source：全量与增量的接缝怎么焊

Source 这一段是整个作业的入口，参数配得对不对，直接决定了“刚上线能不能跑起来”和“跑了三个月稳不稳”。

3.1 版本对齐这张表

我见过最多的启动失败不是配置错，是版本没对上。Flink 主版本、connector 版本、CDC 版本三者之间有个隐含的对应关系，尤其是 Flink 1.14 到 1.15 这段，CDC 2.2 和 2.3 的接口变化不小。

选择原则很简单：以 Flink 版本为主轴，让 CDC 和 ES connector 都去适配它，而不是反过来。因为 Flink 自身的升级成本最高，connector 换一个 jar 就行。

另外有个特别容易忽略的点：CDC connector 是“fat jar”，里面打包了 Debezium 和数据库驱动。如果你把多个版本的 CDC connector 同时丢进lib目录，会出现类冲突，报错信息往往跟实际原因八竿子打不着。我的习惯是lib目录只放当前作业需要的 connector，其他都通过-C参数或者pipeline.classpaths按作业加载。

3.2 Source 参数逐项过

先看 SQL 方式最简版本：

CREATE TABLE order_source ( order_id BIGINT, order_no STRING, user_id BIGINT, mobile_tail STRING, order_status INT, total_amount DECIMAL(18,2), pay_time TIMESTAMP(3), create_time TIMESTAMP(3), update_time TIMESTAMP(3), PRIMARY KEY (order_id) NOT ENFORCED ) WITH ( 'connector' = 'sqlserver-cdc', 'hostname' = '10.0.0.11', 'port' = '1433', 'username' = 'cdc_reader', 'password' = '******', 'database-name' = 'orderdb', 'schema-name' = 'dbo', 'table-name' = 'order_info', 'scan.startup.mode' = 'initial', 'scan.incremental.snapshot.chunk.size' = '8096', 'scan.incremental.snapshot.fetch.size' = '1024', 'connect.timeout' = '30s', 'connect.pool.size' = '8', 'debezium.database.tablename.case.insensitive' = 'false' );

把几个关键项拆开讲。

table-name支持正则，比如order_.*可以一次订阅多张分表。但注意正则匹配是在 schema-name 指定的 schema 下做的，dbo和sales下同名的表不会混淆。多表订阅的时候，并行度就受表数量约束了，因为增量阶段一张表对应一个 reader。你要是订阅了 20 张表但并行度设成 4，那剩下 16 张表只能排队。

scan.incremental.snapshot.chunk.size是全量阶段每次分片读取的行数，默认 8096，这个值是我最喜欢的调优参数之一。设小了，全量阶段频繁查询元数据，慢；设大了，单个分片的查询时间长，checkpoint 容易超时。经验值是 8096 到 16384 之间，表宽（列多列少）差异大的话按行宽调整，宽表往小调，窄表往大调。

scan.incremental.snapshot.fetch.size是 JDBC 的 fetch size，控制每次从游标取多少行回来。设得太小网络往返多，设得太大单批内存占用高。1024 到 4096 之间比较常见，我一般用 1024，稳。

connect.pool.size是连接池大小，默认 20，很多人不动它。但这个值跟作业并行度和全量分片数相关，太大反而会打满源库的连接数上限。我的取值方式是：并行度 × 2，上不超过 20。

debezium.database.tablename.case.insensitive这个参数很值得说。SQL Server 默认是不区分表名大小写的（取决于排序规则），Debezium 默认按不区分处理，也就是Order_Info和order_info都能匹配上。如果你的库里恰好有大小写不同的同名表（虽然不推荐，但老系统里确实有），就会出问题。设成false让它严格匹配，是两个表各同步各的。

3.3 全量到增量的切换逻辑

这是 Flink CDC 2.x 最核心的机制，理解了它，很多“为什么数据重复了”的疑问就自动有了答案。

作业启动后，Source 先做三件事：第一步，给每张表加一个读锁（具体是加锁还是走快照隔离取决于配置），记下当前的 LSN 作为起始位点；第二步，把全量数据按 chunk 切分，并发读取，每个 chunk 读完之后把进度记到 checkpoint 里；第三步，全量读完之后，从那个起始 LSN 开始回放增量变更，并把全量阶段读到的每条记录跟增量流合并去重。

这里有个关键设计：全量阶段读的每条记录，都会拿去和增量流比对 LSN。如果某条记录在全量读取期间被修改过，增量流里会有它的新版本，Flink 会用 LSN 判断哪个更新，保留更新版本。这就是所谓的“无锁读取 + 增量合并”。

这个设计的代价是：全量阶段源库的压力集中在起始那一小段时间，因为要拿 LSN 位点。如果源库上有一张超大表正在跑大批量写入，拿位点可能会等。处理办法是在业务低峰启动，或者先把作业的scan.startup.mode设成latest-offset只跑增量验证链路，确认没问题之后再切回initial补全量。

scan.startup.mode三个可选值的行为要记清楚：

我在生产上踩过一个坑：作业因为 Flink 集群滚动升级停了两小时，恢复的时候用了默认的latest-offset，结果这两小时的变更全丢了，下游 ES 跟源库差了十几万条订单。后来改成作业恢复一律走 checkpoint / savepoint，让 Flink 自己从状态里的位点续上，绝不允许人工指定 latest-offset 重启。这个规矩后来写进了团队的运维手册。

3.4 全量阶段的冲击控制

存量数据大的时候，全量阶段对源库的冲击是实打实的。我参与过的一个环境，订单主表两百多万行，宽表，二十多个列，源库是 4 核 16G 的实例，业务高峰期 CPU 常年 60% 以上。

第一次拉全量，直接把源库 CPU 打到 95%，业务侧开始报警。事后复盘，问题出在三个地方：并行度设了 8（分片并发读）、chunk size 用了默认的 8096（分片多、元数据查询频繁）、没有控制读取速率。

调整之后的做法是这样的：

• 并行度降到 2，让全量读取只占两个读线程。
• chunk size 提到 16384，减少分片数量和元数据交互频次。
• 全量阶段单独安排一个低峰窗口，比如凌晨两点到四点。
• 在源库上开READ_COMMITTED_SNAPSHOT，避免读操作加共享锁阻塞业务写入。

改完之后，全量耗时从“两小时并且把业务打挂”变成了“四十分钟、源库 CPU 峰值 75%、业务无感知”。慢一点但稳，这个交换非常值。

4. 中间层整形：从变更事件到 ES 文档

Flink 作业中间这一层，很多人写得很随意，拿到 CDC 数据直接往 ES 塞。跑起来没问题，但一旦出现问题，排查成本会高得离谱。我倾向于把它当成一个正经的数据整形层来做。

4.1 表到索引的映射策略

最基本的映射是一张表对应一个索引。这个选择在什么情况下会出问题？两种。

第一种是分表合索引。订单按用户 ID 分成了 16 张表，但检索页只有一个，这时候需要把 16 张表的变更汇到一个索引里。做法是用正则订阅多表，然后在 Sink 前用UNION ALL合并，或者用 DataStream 的union。合并的时候要注意各表主键不能冲突，如果用的是自增 ID，两张表的 ID 可能重复，必须在主键里带上分表标识。

第二种是一表拆多索引。同一份数据要给不同的检索场景用，比如运营端要全字段，客户端只要精简字段，这时候用 Flink 的侧输出流或者多次 sink 写到不同索引。我一般倾向于在前端做字段裁剪，也就是同一个作业里INSERT INTO两个 sink 表，各自带不同的字段子集，逻辑清晰。

还有一种是按时间滚动的索引，比如order_2024_06。这种在 ES 里叫时间序列索引，好处是冷热分离、删除旧数据方便。代价是 Sink 的时候要动态计算索引名。Flink 的 ES Sink 支持在 SQL 里用{now()}之类的占位符，但控制粒度有限；如果要按业务的create_time而不是处理时间滚动，就得用 DataStream 自定义IndexGenerator。

4.2 主键与 _id 的生成规则

这是整个链路最关键的一个设计决策，值得单独拎出来。

ES 写入文档有两种方式：不带 _id（自动生成，永远新增，一定重复）、带 _id（同 _id 覆盖，天然幂等）。做 CDC 同步必须选后者，否则每次作业重启都会把数据翻倍。

_id 怎么生成？我列几个方案和它们的适用性。

我的默认选择是用业务层面的唯一键，比如order_no。理由很实在：出问题的时候，你拿到一个 ES 文档，看一眼 _id 就知道是哪个订单，直接去源库查，排查效率高一个数量级。哈希方案虽然优雅，但调试的时候要把 _id 反向算一遍，很别扭。

如果是分表合并场景，就用<分表标识>_<业务主键>的组合。SQL 里可以用拼接：

INSERT INTO order_es SELECT CONCAT(CAST(order_id AS STRING), '_', '01') AS doc_id, order_no, ... FROM order_source;

然后在 Sink 里指定document-id.key-delimiter和主键字段。这里要注意：Flink SQL 的 ES Sink 用主键作为 _id，要求主键字段在 DDL 里声明 PRIMARY KEY，而且只能有一个主键字段（复合主键要用拼接的方式合成一列）。

4.3 类型、时间与空值的处理

类型转换这块，是“看起来没问题、跑起来全是问题”的重灾区。几个高频坑：

DECIMAL 精度丢失。SQL Server 的decimal(18,2)到 Flink 是DECIMAL(18,2)，写 JSON 的时候默认序列化成数字，超过 17 位有效数字就会丢精度。金额字段金额大一点，比如上亿，就会出问题。解决办法是在 Sink 里加：

'json.decimal-as-plain-number' = 'true'

这个配置让 decimal 序列化成字符串而不是浮点数，ES 侧 mapping 里对应字段设成scaled_float或者keyword。我一般用scaled_float+scaling_factor = 100，既能范围查询又不丢精度。

时间类型时区错乱。SQL Server 的datetime没有时区信息，Flink 读出来按TIMESTAMP处理，写 JSON 的时候默认格式跟服务器时区相关。如果 Flink 集群和 SQL Server 在同一个时区还好，跨时区部署就全乱了。我的做法是统一加配置：

'json.timestamp-format.standard' = 'ISO-8601'

并且约定所有时间字段在 ES 里映射成date类型，格式strict_date_optional_time||epoch_millis。这样不管哪种表示方式，ES 都能解析。

空值与字符串 "null"。SQL Server 里的 NULL 到 Flink 是 null，写 JSON 的时候默认会省略这个字段（ES 里就是没有这个 field），而不是写成 null。大多数场景这是对的。但要小心一种情况：源库里有人把 NULL 手动改成了字符串'null'或者空串，这种值会原样同步过去，检索的时候就有两种“空”并存。我在整形层加了一个CASE WHEN把空串统一转成 null：

CASE WHEN order_remark = '' THEN NULL ELSE order_remark END AS order_remark

顺便说一句，上面这些整形逻辑，我建议全放在 SQL 里做，不要写自定义 UDF。SQL 的方式可读性好、改动成本低，出问题的时候一眼就能看出哪一步做错了。UDF 一旦多了，作业就变成了一个黑盒，接手的人会很痛苦。

5. ES Sink：批量、重试与幂等

Sink 这一段看起来最简单，往 ES 写嘛。但生产环境跑久了会发现，绝大多数数据不一致的问题，根子都在 Sink 的批量和重试配置上。

5.1 Sink connector 的选择

Flink 官方提供的 ES connector 有elasticsearch-6和elasticsearch-7两个，7 这个通常兼容 7.x 和 8.x 的集群（8.x 需要注意是否开启了安全认证）。选型上没什么纠结的，新项目一律用 elasticsearch-7。

关键配置项如下：

CREATE TABLE order_es ( doc_id STRING, order_no STRING, order_status INT, total_amount STRING, pay_time TIMESTAMP(3), update_time TIMESTAMP(3), PRIMARY KEY (doc_id) NOT ENFORCED ) WITH ( 'connector' = 'elasticsearch-7', 'hosts' = 'http://10.0.0.21:9200;http://10.0.0.22:9200', 'index' = 'order_info_v1', 'document-id.key-delimiter' = '$', 'sink.bulk-flush.max-actions' = '2000', 'sink.bulk-flush.max-size' = '10mb', 'sink.bulk-flush.interval' = '1000', 'sink.bulk-flush.backoff.strategy' = 'EXPONENTIAL', 'sink.bulk-flush.backoff.max-retries' = '8', 'sink.bulk-flush.backoff.delay' = '1000', 'sink.max-retries' = '5', 'format' = 'json' );

hosts里配多个节点，用分号隔开。至少要配两个，形成冗余，避免单个节点挂掉导致整个 Sink 不可用。

index我习惯带上版本号_v1。因为 ES 的 mapping 一旦建了就很难改（改字段类型需要 reindex），版本号让你能在需要时平滑切到_v2，用别名切换，不需要停机。

5.2 批量参数怎么算出来

这三个bulk-flush参数是最需要算的。

sink.bulk-flush.max-actions是单批最多多少条文档，max-size是单批最大字节数，interval是多久强制刷一次。三个条件是“或”的关系，先到先触发。

计算思路是从 ES 侧的承受能力倒推。假设你的 ES 集群是 3 个数据节点，每节点 8 核，索引有 6 个主分片。大量写入的时候，单个 bulk 请求如果打到 2000 条文档、每条约 1KB，就是 2MB 一个请求。这个体量对 ES 来说很舒适，单个请求通常在 100ms 内返回。

再算吞吐：checkpoint 间隔假设 30 秒，作业并行度 4，每个并行实例每秒产生约 500 条变更（这是一个中等的写入量），那么 30 秒内单并行实例积累 15000 条，会触发 7 到 8 次 flush。这个量级下，max-actions = 2000是合理的。

反过来，如果你的数据量很小，每小时才几百条变更，那max-actions设再大也没意义，因为永远是interval先触发。这时候把interval从 1000 降到 500 更好，延迟更低，代价是请求数变多。反过来数据量特别大，max-actions可以提到 5000，但要同时看max-size，别让单个请求超过 10MB。

这里给个我常用的对照表：

这张表不是定理，是起点。真实取值要看单条文档的平均大小和 ES 集群规格，上线前用压测数据调一轮，比什么都靠谱。

5.3 幂等与一致性：ES 没有事务，靠什么兜底

这是整篇文章里最需要说透的一点，也是最容易被误解的。

Flink 的 ES Sink 提供的是at-least-once语义：checkpoint 完成的时候，已经提交的 bulk 请求是确定生效的，但 checkpoint 之间失败重启，那段时间的请求会重放。ES 支持按 _id 覆盖，所以重放不会产生重复文档，结果是“至少写一次，但内容一致”。

很多人看到这里会问：这算不算 exactly-once？严谨的说法是：写入次数上不是一次性，但最终数据状态是准确的。对于检索场景，这个语义完全够用，因为用户看到的永远是最终状态。对于账务类场景，那就需要下游再做一层幂等校验，比如按版本号或时间戳比较，旧的覆盖新的要拒绝掉。

这里有个非常隐蔽的顺序问题：同一个文档在短时间内被更新多次，重放之后可能不是最新版本。比如订单状态在 30 秒内从 1 改成 2 又改成 3，如果没有 checkpoint 就重启，重放的时候可能 2 那条后到，把 3 覆盖了。要防这个，要么让 ES 侧做版本控制（用version字段），要么在整形层加一个update_time字段，写入前通过 ES 的_updateAPI 带脚本做“只在新版本更新时覆盖”。Flink SQL 的 ES Sink 不支持这种脚本，得上 DataStream 自定义请求工厂。

我的经验是：大多数团队不需要这么复杂，但你必须知道这个风险存在。如果业务方问“会不会丢数据、会不会错数据”，你要能给出准确的答案，而不是一句“Flink 保证精确一次”糊弄过去。

5.4 索引侧的前置准备

Sink 之前，ES 那边要做好准备，否则会有一堆烦人的小问题。

第一，先建索引和 mapping，别依赖自动映射。自动映射对字符串默认用text+keyword双字段，对数字会推断成long或float。金额字段被推断成float就是灾难。我的做法是提前用 PUT 建好：

curl -X PUT "http://10.0.0.21:9200/order_info_v1" -H 'Content-Type: application/json' -d ' { "settings": { "number_of_shards": 6, "number_of_replicas": 1, "refresh_interval": "5s" }, "mappings": { "properties": { "order_no": { "type": "keyword" }, "order_status": { "type": "integer" }, "total_amount": { "type": "scaled_float", "scaling_factor": 100 }, "pay_time": { "type": "date" }, "update_time": { "type": "date" }, "order_remark": { "type": "text", "analyzer": "standard" } } } }'

分片数怎么定？有个粗算公式：分片数 ≈ 并行度 × 1.5 到 2，同时考虑单分片大小控制在 10G 到 50G 之间。6 个分片配 4 个并行度，是我用得最多的组合。

refresh_interval设成5s而不是默认的1s，是因为批量写入时频繁 refresh 会吃掉大量 CPU。检索场景 5 秒的可见延迟基本感知不到。如果对实时性要求极高，那就保持 1s，但要做好 CPU 涨 20% 到 30% 的准备。

第二，写入账号的权限要控制住。同步账号只需要目标索引的write和create_doc权限，不要给manage，避免一个配错把整个索引删了。这种事我听说过不止一次。

6. 上线后的问题排查实录

下面这些问题都是我或者身边同事真实遇到过的，按出现频率排。

6.1 启动就报错的几类

第一类：The connector does not support the table或者表找不到。

九成是schema-name写错。SQL Server 里默认 schema 是dbo，但老系统里可能有sales、order这种自定义 schema。排查方法：

SELECT s.name AS schema_name, t.name AS table_name FROM sys.tables t JOIN sys.schemas s ON t.schema_id = s.schema_id WHERE t.name = 'order_info';

顺便确认一下表是不是真的开启了 CDC：

SELECT * FROM cdc.change_tables WHERE source_object_id = OBJECT_ID('dbo.order_info');

查不到就是没开，或者开了又被禁用了。

第二类：Cannot find the capture instance。

这个报错的意思是 Debezium 要找的 capture instance 名对不上。SQL Server 侧捕获实例名默认是<schema>_<table>，如果当初开启时自定义了名字，或者表名里有特殊字符被转义了，就会失败。查一下实际名字：

SELECT capture_instance FROM cdc.change_tables;

第三类：权限不足，SELECT permission was denied on the object 'change_tables'。

同步账号需要cdcschema 的 SELECT 权限。最小授权方式：

GRANT SELECT ON SCHEMA::cdc TO cdc_reader;

注意cdcschema 下只有部分对象是可查询的，直接 GRANT 到 schema 级别最省事。

第四类：JDBC 驱动问题。SQL Server 的驱动 jar 有多个版本，mssql-jdbc9.x 及以上需要 JDK 8u191 或更高。驱动版本和 JDK 不匹配的时候会报很奇怪的类找不到错误。建议看下 Flink 的lib目录里有没有多个版本的驱动，只留一个。

6.2 跑起来之后才暴露的

第一类：ES 报es_rejected_execution_exception。

这说明 ES 的写入线程池队列满了，通常是因为 bulk 请求太大或者太频繁。处理思路按优先级：

1. 降低max-actions和max-size，让单批请求更小。
2. 增加backoff.max-retries，让失败请求有更多重试机会。
3. 从集群侧看，是不是有节点负载不均（分片分配问题），或者磁盘 IO 打满。

我在一个环境里遇到过，单纯调小批量没用，最后发现是索引的分片分布不均，6 个分片有 5 个在同一个节点上。重建索引指定了路由才解决。

第二类：同步延迟越来越大。

这个现象很有迷惑性，因为作业不报错，只是越来越慢。排查顺序是：先看 Flink UI 的 source 端是不是有反压（backpressure），反压说明下游处理不过来；如果 source 没反压，说明问题在增量读取侧，去看变更表的增长速度和读取速度是否匹配。

有一个很常见的原因是源库变更表清理作业跑得太频繁或者太慢。跑得频繁会把还没读的 LSN 清掉，导致读取中断；跑得慢会让变更表变大，扫描变慢。要观察清理作业的执行记录：

EXEC sys.sp_cdc_help_jobs; SELECT * FROM sys.dm_cdc_log_scan_sessions ORDER BY start_time DESC;

第二句能看到每次日志扫描的行数、耗时、错误。

第三类：checkpoint 持续失败。

checkpoint 超时常见于几种情况：状态太大（RocksDB 的 checkpoint 时间长，要开增量 checkpoint）、下游 ES 响应慢导致 barrier 对齐卡住、单次 bulk 请求卡住不返回。

第二条最容易忽略。ES 的某个节点如果因为磁盘满或者 GC 卡顿不响应，bulk 请求会一直挂着直到超时，这期间 barrier 无法对齐，checkpoint 就一直失败。解决办法是把sink.bulk-flush.backoff.delay设得合理，并且给 Flink 的execution.checkpointing.timeout留足余量，比如 10 分钟。

6.3 数据对不上的排查路径

这是最棘手的一类，因为不报错，只是数字对不上。

我的固定排查顺序是这样的：

第 4 步的查询方式：

curl -s "http://10.0.0.21:9200/order_info_v1/_search" -H 'Content-Type: application/json' -d ' { "size": 0, "aggs": { "dup_check": { "cardinality": { "field": "_id", "precision_threshold": 40000 } } } }'

如果 _id 是唯一的但总行数比源库多，那基本就是“删除事件没处理”。如果行数少，就看增量流是不是从中途开始读的，或者是变更表被清理掉了。

我印象最深的一次：ES 里比源库多了三百多条，查了半天发现是上游的删除操作用的是逻辑删除（is_deleted = 1），而 Flink 作业只处理了物理删除（op = d），逻辑删除的记录在 ES 里一直留着。后来在 Sink 前加了一个过滤：

INSERT INTO order_es SELECT ... FROM order_source WHERE is_deleted = 0;

看起来简单，但如果你没意识到“逻辑删除也算删除”，就会一直以为是 CDC 丢数据。

提醒一句：用WHERE过滤掉逻辑删除的记录，意味着 ES 里对应的文档不会被删，只会永远停在上一次的状态。正确做法是转发一个标记字段到 ES，让检索端自己过滤，或者用op = u AND is_deleted = 1的时候发一个物理删除。这个细节很容易做错，而且做错之后很难发现。

7. 一些踩过坑之后沉淀的做法

写到最后，分享几条我在实际项目里验证过、也确实吃过教训的做法，不是教科书上的原则。

关于全量阶段的开关。我给很多项目加了一个“全量模式”的启动参数：首次上线走initial，之后所有重启一律走 checkpoint 恢复，绝不允许latest-offset。为了防止误操作，作业的启动脚本里对启动模式做了校验，非首次启动如果检测到initial，直接拒绝启动并告警。这个看起来麻烦的规矩，避免过一次严重事故：有人在生产误触了重刷全量，那是一次两百多万行的读，源库差点扛不住。

关于SQL 与 DataStream 的取舍。链路简单、只做字段映射和过滤，就用 SQL，开发快、改起来快。涉及跨表关联、动态索引名、复杂的条件删除，就老老实实上 DataStream。我见过用纯 SQL 硬凑复杂逻辑的作业，三百多行 SQL，接手的人看了两天才看明白，最后推倒重写只花了一天。

关于监控指标。除了 Flink 自带的numRecordsIn/numRecordsOut，我建议额外打三个业务指标：每秒处理的变更条数、ES bulk 请求的平均响应时间、当前处理位点与源库最新 LSN 的差值（延迟）。第三个指标最能反映真实同步延迟，比 Flink UI 的 lag 更直观。实现方式是在作业里起一个定时器，周期性查一下cdc.lsn_time_mapping的最大 LSN 对应时间，跟当前处理的时间做差。

关于回滚方案。这条链路上线前，一定要想好“如果 ES 写脏了怎么办”。我的做法是索引带版本号，写入走别名，出问题把别名切回_v0，五分钟完成回滚，然后离线慢慢修数据。别指望在线上直接改 ES 里的文档，几千条还能脚本处理，几十万条就是个灾难。

还有一个很小的习惯，但收益很大：在 ES 文档里加三个元字段——_src_update_time（源库更新时间）、_sync_time（同步写入时间）、_sync_batch（批标识）。这三个字段平时用不上，但排查“为什么这条数据是旧的”的时候，看一眼就知道是源库没更新、还是同步延迟、还是重放覆盖。加字段的成本几乎为零，省下的排查时间是以小时计的。