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 开启数据库级和表级 CDCAgent 作业把事务日志扫进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 这套组合的能力边界这里需要先立个边界免得大家做完才发现预期不对。能力项是否支持说明行级增删改捕获支持insert/update/delete 分别对应 op c/u/d秒级延迟支持取决于 checkpoint 间隔和 ES refresh 策略存量数据初始化支持scan.startup.mode initial会先做全量快照结构变更DDL自动同步有限SQL Server CDC 对 DDL 跟踪弱加列需重启作业ES 侧事务回滚不支持ES 无事务只能靠主键覆盖做到最终一致跨库多表任意关联部分单表同步稳多表 join 后写入要自己控制顺序超大表全量需分片靠chunk.size和并行度控制否则容易把源库读爆其中“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_mappingLSN 和时间戳的映射增量位点就是靠它换算的。cdc.captured_columns/cdc.index_columns元数据。cdc.dbo_表名_CT真正的变更表每张被跟踪的表对应一张。变更表里除了原表的全部列还有五个元数据列这些列是理解 CDC 的关键列名含义排查时怎么用__$start_lsn事务提交的 LSN同一事务的多行相同判断变更顺序、定位断点__$seqval同一事务内的序号同一事务内多行排序用__$operation1删除 2插入 3更新前镜像 4更新后镜像排查“为什么 update 变成两条”__$update_mask哪几列被更新了做字段级同步时用__$command_id命令序号极少用到排查乱序时看一眼这里有个特别容易踩的点__$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 Ndbo, source_name Norder_info, role_name NULL, capture_instance Ndbo_order_info, supports_net_changes 1, filegroup_name NCDC;逐个说下这些参数为什么这么选。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_datareaderdb_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 作业挂了两天回来还能不能续上”。如果你的作业可能因为发布、故障停超过一天把这个值调到 100807 天更稳妥EXEC sys.sp_cdc_change_job job_type Ncleanup, 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说明1.13.x2.1.xelasticsearch-6/7版本较老建议升级1.14.x2.2.x ~ 2.3.xelasticsearch-7稳定性不错广泛使用1.15.x2.3.x ~ 2.4.xelasticsearch-7推荐组合1.16.x 及以上2.4.x 及以上elasticsearch-7新项目优先选这个区间选择原则很简单以 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三个可选值的行为要记清楚模式行为适用场景initial先全量再增量首次上线、补齐历史latest-offset跳过全量从当前位点开始验证链路、故障恢复后不想重刷earliest-offset从变更表最早位点开始补数但要注意保留窗口够不够我在生产上踩过一个坑作业因为 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_id简单直接分表合并时冲突组合键order_id 分表号分表场景必需长度略长哈希md5(order_id)定长适合超长主键不可读排查困难业务唯一键order_no可读跨系统唯一依赖业务键稳定我的默认选择是用业务层面的唯一键比如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_floatscaling_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把空串统一转成 nullCASE 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。这里给个我常用的对照表日变更量级max-actionsmax-sizeinterval10 万以内10005mb500ms10 万 ~ 100 万200010mb1000ms100 万 ~ 1000 万300015mb1500ms1000 万以上500020mb2000ms这张表不是定理是起点。真实取值要看单条文档的平均大小和 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别依赖自动映射。自动映射对字符串默认用textkeyword双字段对数字会推断成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;顺便确认一下表是不是真的开启了 CDCSELECT * 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 请求太大或者太频繁。处理思路按优先级降低max-actions和max-size让单批请求更小。增加backoff.max-retries让失败请求有更多重试机会。从集群侧看是不是有节点负载不均分片分配问题或者磁盘 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 数据对不上的排查路径这是最棘手的一类因为不报错只是数字对不上。我的固定排查顺序是这样的步骤检查项判断标准1源库和 ES 的总行数对比差异应该为 0排除逻辑删除的话会一致2ES 索引的 refresh 是否完成查之前手动 refresh避免误判3是否有op d的事件被丢弃看作业日志里是否有过滤逻辑4是否有重复 _id用 cardinality 聚合不重复则正常5抽样比对 10 条记录的字段逐字段核对定位到具体字段6检查位点是否回退看作业重启记录和 checkpoint 时间第 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批标识。这三个字段平时用不上但排查“为什么这条数据是旧的”的时候看一眼就知道是源库没更新、还是同步延迟、还是重放覆盖。加字段的成本几乎为零省下的排查时间是以小时计的。