简介：面向使用 FlinkCDC 接入达梦数据库、构建日志级实时同步链路的开发与数据工程师，这份工程资料聚焦达梦变更数据捕获在 Flink 中的落地实践，覆盖连接参数配置、变更事件反序列化、流式处理与结果统计等关键环节，也适合需要对照 Java 作业与 Flink SQL 两种同步写法的读者学习。压缩包共 315 个文件，约 341.71MB，以 jar 依赖包、java 与 class 示例、xml 与 properties 配置、sql 脚本及 docx 说明为主，另含少量构建与工程配置文件，便于还原运行环境、核对依赖版本并梳理项目结构。已有 776 人学习下载，反映该主题在实时数据同步场景中关注度较高。读者可从中获得达梦 CDC 接入的参考实现，理解日志捕获如何降低对源库性能的影响，并借鉴反序列化、统计工具与测试用例的组织方式，为实时报表、数据监控和告警等事件驱动应用提供落地思路。

1. 为什么 FlinkCDC 撞上达梦，第一件事不是写代码

FlinkCDC 接达梦数据库做基于日志的实时同步，最容易出错的地方不是代码写不出来，而是很多人默认它像 MySQL 那样引个依赖就能跑——翻遍 Flink CDC 的官方连接器清单，你找不到一个叫 dameng 的 artifact。这意味着从达梦到 Flink 的这段链路需要自己搭建：一侧是达梦的联机日志和归档日志，另一侧是 Flink CDC 认得的那种带 before/after 的变更事件流，中间缺的那块日志解析和位点管理，得自己补上。适合读这篇的人有三类：手上有一套达梦在跑核心业务、下游要进 Flink 做实时宽表或入湖的；被时间戳增量同步坑过、想换成日志方案的；以及已经拿到某个达梦同步组件、但不清楚它和 Flink 之间该怎么接的。下面按我实际拆这条链路的顺序讲。

2. 拆开链路：从达梦日志到 Flink 变更流的四段结构

在动手之前先把结构想清楚，后面写代码会省很多返工。整条链路其实是四段：达梦产生日志 → 有人把日志解析成行级变更 → 变更被序列化成统一事件格式 → Flink 消费并按位点做容错。四段里最容易出问题的是第二段和第四段，因为第一段是数据库自己的事，第三段只要格式定死了就很少改。

我见过不少团队一上来就写 Flink 作业，结果解析端的行为没定清楚，位点一会儿用文件名一会儿用偏移量，最后对不上账，只能整条链路重来。所以这一章先把几个必须先想明白的选择摊开。

2.1 时间戳增量为什么在达梦上特别容易出错

最省事的同步方式是用UPDATE_TIME字段做轮询，很多团队的第一版都是这么写的，因为它不需要动数据库配置，一个 JDBC 连接加一条 SQL 就能跑。但这条路在真实业务表上会陆续暴露问题。

-- 典型的轮询增量写法，看着没问题 SELECT id, name, amount, update_time FROM biz_order WHERE update_time >= TO_DATE('2024-06-01 10:00:00','yyyy-mm-dd hh24:mi:ss') ORDER BY update_time;

用>=会让上一批的最后一条重复消费，改成>又会漏掉同一秒里写入的多条记录。更麻烦的是物理删除根本捕获不到，DELETE之后源表少了一行，轮询方案永远不知道要同步这个动作。有些团队用软删除绕过去，但下游宽表里那条记录会一直挂着，没人敢真的相信它。

还有一个隐性问题是精度。达梦里时间字段的精度取决于建表时的定义和系统参数，秒级字段配上高频写入的业务，一个批次里同一秒的记录能有几百条，靠时间戳排序本身就不稳定。再加上批量更新时应用可能把整批记录写成同一个SYSDATE，你拿这个字段做游标，等于把几百条记录挤在一个点上反复扫。

真正要命的是拿不到前像。轮询只能看到更新后的值，宽表里做累加、做状态回撤、做金额变更对比，全都做不了。日志方案之所以值得折腾，核心价值就在这个 before 值上——它让下游可以重放一次变更，而不只是接收一个最终状态。

2.2 两条日志路线怎么选：自己解析还是用现成组件

达梦的日志解析有两条路，选哪条取决于你的表数量和维护人力，不取决于技术上哪个更优雅。

我的判断标准比较粗暴：表数量在几十张以内、变更量每秒几百条、团队里有人愿意维护解析代码，走第一条；表上百张、业务方还在频繁改表结构、没人愿意半夜被 DDL 炸醒，走第二条，把日志捕获交给组件，Flink 只负责消费和计算。

两条路共同的落点是消息队列。中间放一层 Kafka 不是为了架构好看，是为了解耦和回放：解析端挂了不影响 Flink 作业，Flink 作业改逻辑要重跑历史数据的时候，也不用去数据库重新挖日志。这个决定在后期会救你很多次。

2.3 变更事件的 schema 必须先定死再写代码

格式没定就写代码，最后一定会在字段命名和空值语义上返工。我一般直接对齐 Debezium 的 envelope 结构，因为 Flink 生态里大量现成的反序列化器和下游 sink 都认这套，省得自己再写一层转换。

{ "op": "u", "before": {"ID": 1001, "AMOUNT": 12.50}, "after": {"ID": 1001, "AMOUNT": 30.00}, "source": { "db": "BIZ", "table": "BIZ_ORDER", "lsn": "00000000000a2f31", "ts_ms": 1717214400123 }, "ts_ms": 1717214400456 }

几个字段的语义要在团队内约定死：op用小写，c插入、u更新、d删除、r快照读，这样和 Debezium 完全对齐；before在插入时为 null，after在删除时为 null，不要用空对象代替，下游判断会出错；source.lsn存成十六进制字符串而不是数字，因为前端 JavaScript 的 Number 只有 53 位有效精度，用 Long 传过去会静默丢位，这个坑我踩过一次，排查了一整晚。

source.ts_ms和外面的ts_ms是两个东西，前者是这条变更在数据库里发生的时间，用日志里的时间戳换算；后者是解析端处理它的时间。两个都留着，端到端延迟就是用后者减前者算出来的。少了任何一个，线上延迟升高的时候你只能干瞪眼。

lsn用 CSCN 而不是归档文件偏移，这一点在第 4 章细讲，这里先记住结论：归档文件会被切、会被清理，库内的 CSCN 是单调的，拿它当全局位点才稳。

2.4 开工前的环境与权限清单

动手前把这张表逐项确认一遍，比写到一半发现权限不够要快得多。

提示：达梦不同版本里部分视图和参数的字段名会有差异，上面提到的东西不要照抄，先SELECT * FROM V$XXX看一眼实际字段再写 SQL，这一步花不了两分钟。

权限这块值得多说一句。日志解析需要读 redo 内容，普通业务账号是读不到的，常见做法是单独建一个同步专用账号，把日志解析包的执行权限和几个日志视图的查询权限单独授出去，而不是直接把 DBA 给出去。给出去容易，收回来难，尤其是这套东西后面要交接给别人的时候。

3. 把达梦日志读成变更事件：三步跑通最小闭环

这一章的目标很明确：不写 Flink 作业，先在命令行把「一条 UPDATE 从达梦日志里被解析成一行 JSON」这件事跑通。这一步能通，后面的工程化都是体力活；这一步不通，后面写多少代码都是空中楼阁。

3.1 开归档并确认起始位点

达梦默认大概率没开归档，而没有归档就没有可回溯的日志，同步无从谈起。切换归档模式需要实例短暂进入 MOUNT 状态，期间数据库不可用，所以必须在业务低峰做，并且提前打好招呼。

-- 1) 先确认现状：归档目录在哪、有没有归档产生 SELECT * FROM V$DATABASE; SELECT * FROM V$ARCHIVED_LOG ORDER BY FIRST_TIME DESC; -- 2) 未开归档时切换（需要 SYSDBA）：MOUNT 期间库不可用 ALTER DATABASE MOUNT; ALTER DATABASE ARCHIVELOG; ALTER DATABASE OPEN; -- 3) 记录解析起点位点，字段名以实际查询结果为准 SELECT * FROM V$RLOG;

第一步的两个查询不只是确认状态，还能让你知道归档目录在哪个盘、最近的归档是什么时候产生的。如果V$ARCHIVED_LOG是空的，说明要么没开归档，要么开了但归档目录配置有问题，后者更隐蔽——看着归档模式是开的，实际一个文件都没落地。

第二步的三条语句必须连着执行，中间隔太久会影响业务。有些环境不允许这么切，那就走配置文件的方式，在归档配置里打开开关并指定目录，重启实例生效。

# dmarch.ini 片段：本地归档 [ARCHIVE_LOCAL1] ARCH_TYPE = LOCAL ARCH_DEST = /dmdata/arch ARCH_FILE_SIZE = 1024 ARCH_SPACE_LIMIT = 20480

ARCH_FILE_SIZE单位是 MB，单个归档文件多大切一个；ARCH_SPACE_LIMIT是归档空间上限，单位 MB，0表示不限制。生产上千万不要留0，磁盘写满导致数据库挂起的场面我见过一次，恢复起来比配置它麻烦十倍。设了上限之后要配套清理策略，清理的时机必须晚于解析端消费到的位点，否则会出现「解析端要读的文件已经被删了」这种断点丢失。

第三步拿到的位点就是这条链路的初始游标，记下来，后面所有代码都从它开始。不同版本的V$RLOG字段不一样，我一般直接SELECT *把整行打出来，找到那个表示当前写入位置和检查点位置的字段，通常是带LSN字样的。

3.2 用 DBMS_LOGMNR 解析一段归档日志

达梦提供了和 Oracle 那套高度相似的日志解析包，用法基本一致：先添加日志文件，再启动解析会话，然后查内容视图，最后显式结束。

-- 添加要分析的归档日志，NEW 表示新开一个解析会话 BEGIN DBMS_LOGMNR.ADD_LOGFILE( LOGFILENAME => '/dmdata/arch/ARCHIVE_LOCAL1_0x0001_20240601.log', OPTIONS => DBMS_LOGMNR.NEW); END; / -- 用在线字典启动解析，表结构没变过的情况下够用 BEGIN DBMS_LOGMNR.START_LOGMNR( OPTIONS => DBMS_LOGMNR.DICT_FROM_ONLINE_CATALOG); END; / -- 只看关心的表，按日志内的定位字段排序保证事务内顺序 SELECT OPERATION, SEG_OWNER, TABLE_NAME, RS_ID, SSN, CSCN, SQL_REDO, SQL_UNDO, TIMESTAMP FROM V$LOGMNR_CONTENTS WHERE TABLE_NAME = 'BIZ_ORDER' ORDER BY RS_ID, SSN; -- 用完必须结束，否则会话一直挂着占资源 BEGIN DBMS_LOGMNR.END_LOGMNR; END; /

OPERATION会给出INSERT、UPDATE、DELETE，还有COMMIT和 DDL 类的记录，这些都要处理，不能只挑三种 DML 看。RS_ID加SSN是日志内的定位坐标，同一事务的记录靠它排序才是原始顺序。CSCN是事务提交时的系统变更号，用它判断「这个事务到底提交了没有」——日志里出现的记录不代表事务成功，回滚的事务也会留痕，这一点在第五章会展开说。

SQL_REDO是重做语句，能还原出变更后的值；SQL_UNDO是回滚语句，UPDATE和DELETE的前像就藏在它的 WHERE 条件里。这是整个方案能拿到 before 值的根本原因，也是它比时间戳轮询强的地方。

DICT_FROM_ONLINE_CATALOG有个前提：表结构从产生日志到现在没变过。一旦中间加过列、改过类型，解析出来的列名和值就会错位。这种情况要用从日志里重建字典的方式，代价是慢很多。我的经验是表结构变更频繁的库，别硬扛这条路，直接上现成同步组件。

3.3 把解析结果归一化成 Debezium 风格 JSON

拿到SQL_REDO和SQL_UNDO只是半成品，还得把它们还原成结构化的字段值。这段代码是整个链路里我最不愿意重写第二遍的部分，因为字符串解析的边界情况太多了。

import re, json, time OP_MAP = {"INSERT": "c", "UPDATE": "u", "DELETE": "d"} def _cast(v): """把日志里的字面量转成 Python 值，字符串去引号，数字转数值""" v = v.strip().rstrip(";") if v.upper() == "NULL": return None if v.startswith("'") and v.endswith("'"): return v[1:-1].replace("''", "'") # 转义的单引号还原 try: return float(v) if "." in v else int(v) except ValueError: return v # 函数、表达式一律原样保留 def parse_after(sql_redo): """从 SQL_REDO 还原变更后的值 INSERT: INSERT INTO "BIZ"."BIZ_ORDER"("ID","AMOUNT") VALUES(1001,30.0); UPDATE: UPDATE "BIZ"."BIZ_ORDER" SET "AMOUNT"=30.0 WHERE "ID"=1001; """ if not sql_redo: return None up = sql_redo.upper() if up.startswith("INSERT"): cols = re.search(r"\((.*?)\)\s*VALUES", sql_redo, re.S | re.I) vals = re.search(r"VALUES\s*\((.*)\)", sql_redo, re.S | re.I) if not cols or not vals: return None names = [c.strip().strip('"') for c in cols.group(1).split(",")] values = [_cast(v) for v in vals.group(1).split(",")] return dict(zip(names, values)) if up.startswith("UPDATE"): m = re.search(r"\bSET\b(.*?)\bWHERE\b", sql_redo, re.S | re.I) if not m: return None return {k: _cast(v) for k, v in re.findall(r'"(\w+)"\s*=\s*([^,]+)', m.group(1))} return None def parse_before(sql_undo): """从 SQL_UNDO 的 WHERE 条件还原变更前的值，只处理等值条件""" if not sql_undo: return None parts = sql_undo.split("WHERE", 1) if len(parts) < 2: return None out = {} for k, v in re.findall(r'"(\w+)"\s*=\s*(.+?)(?:\s+AND\s+|$)', parts[1], re.I): out[k] = _cast(v) return out def to_event(row, db_name): op = OP_MAP.get(row["OPERATION"]) if op is None: # COMMIT / DDL 单独走分支 return None return { "op": op, "before": parse_before(row["SQL_UNDO"]) if op in ("u", "d") else None, "after": parse_after(row["SQL_REDO"]) if op in ("c", "u") else None, "source": { "db": db_name, "table": row["TABLE_NAME"], "lsn": format(int(row["CSCN"]), "016x"), # 16 进制定长，方便字符串比较 "ts_ms": int(row["TIMESTAMP"].timestamp() * 1000), }, "ts_ms": int(time.time() * 1000), }

几个设计取舍要说清楚。format(int(CSCN), "016x")补零到 16 位，是为了让位点在字符串字典序下就等于数值序，后面做增量拉取时可以直接用lsn > last_lsn比较，不用来回转类型。parse_before只处理等值条件，碰到WHERE "AMOUNT" > 100这种就放弃并打日志——日志解析不是 SQL 解释器，硬要全覆盖只会写出一个更难维护的东西，分场景处理更实际。

TIMESTAMP字段的时区一定要确认。达梦返回的是数据库本地时间，Flink 作业那边如果默认按 UTC 解释，算出来的端到端延迟会整整差一个时区。我一般在这层就统一成 UTC 毫秒，全链路只认这个，展示层再转回本地。

3.4 先落 Kafka，再让 Flink 来消费

解析端和 Flink 之间隔一层 Kafka，理由前面说过：解耦、回放、多下游。建 topic 的时候有两个参数要特别留意。

# 按主键 hash 到同一分区，保证同一行的变更严格有序 kafka-topics.sh --bootstrap-server broker:9092 \ --create --topic dm-biz-order-cdc \ --partitions 3 --replication-factor 2 \ --config cleanup.policy=compact \ --config retention.ms=604800000

cleanup.policy=compact会为每个 key 保留最后一条消息，这正好是 CDC 场景需要的语义：下游重建状态的时候只关心每行的最新值。消息的 key 用库名.表名.主键值拼，这样同一行的所有变更必然落到同一分区，顺序有保证。分区数按峰值吞吐来定，3 个分区在每秒几千条变更的量级够用，别一上来就开 20 个，消费者并行度跟不上反而增加管理成本。

Flink 侧先用 PyFlink 跑一个最简的消费作业，确认能从 topic 里读出数据并打到标准输出，再往里加业务逻辑。

from pyflink.datastream import StreamExecutionEnvironment from pyflink.datastream.connectors.kafka import KafkaSource, KafkaOffsetsInitializer from pyflink.common.serialization import SimpleStringSchema from pyflink.common import WatermarkStrategy env = StreamExecutionEnvironment.get_execution_environment() env.enable_checkpointing(30_000) # 30 秒一次，CDC 场景别开太长 src = KafkaSource.builder() \ .set_bootstrap_servers("broker:9092") \ .set_topics("dm-biz-order-cdc") \ .set_group_id("flink-dm-cdc") \ .set_starting_offsets(KafkaOffsetsInitializer.committed_offsets()) \ .set_value_only_deserializer(SimpleStringSchema()) \ .build() ds = env.from_source(src, WatermarkStrategy.no_watermarks(), "dm-cdc-source") ds.map(lambda s: s).print() # 这里换成你的 upsert / 建宽表逻辑 env.execute("dm-cdc-to-flink")

enable_checkpointing(30_000)这个 30 秒直接决定故障恢复时最多回多少数据。CDC 场景下 checkpoint 间隔太长，恢复时重复消费的量就大；太短又会让整个作业频繁做快照，影响吞吐，30 秒是个比较常见的折中。committed_offsets()依赖 checkpoint 完成才提交位点，所以作业崩了之后会从上一次成功的快照续上，不会从头读。

WatermarkStrategy.no_watermarks()是有意为之。CDC 流里的记录天然乱序——不同表、不同事务的日志混在一起到达，水位线必须基于source.ts_ms单独定义，直接给个默认的反而会让窗口计算提前触发、结果偏小。等下游真要开窗口的时候，再按业务时间显式指定。

4. 自定义 Source：位点、快照与精确一次这三件事

跑到上一章结束，你已经有一条能用的链路了，只是它还是「先解析成 JSON 再投递」的两段式。表少、量小的时候这样够用，一旦表多起来或者想减掉中间那层，就得把解析逻辑收进 Flink 的 Source 里，让位点管理交给 Flink 的 checkpoint 机制去做。这一章讲的就是这一步的三个关键决定。

4.1 位点选型：归档文件偏移、LSN 还是 CSCN

三种位点我都用过，结论是只有一种适合当全局游标。

归档文件加文件内偏移是最直观的，但归档文件会被切分、被清理，你记下的「第 17 个文件的第 4096 字节」在下一次故障恢复时很可能已经不存在了。用它做位点，等于把恢复能力绑死在归档保留策略上。

日志序号是库内单调递增的，比文件偏移稳，但它的粒度和事务边界不一定对齐，用它做断点续传时容易出现「从半个事务中间开始读」的情况，得配合事务号做二次过滤。

CSCN 是事务提交时的系统变更号，天然带事务边界语义，而且库内单调。用它当全局位点，恢复时直接按CSCN > last_cscn拉取，天然不会从半个事务中间切进去。代价是需要额外维护一个「同一 CSCN 下多条记录」的去重逻辑，因为一个事务里可能有多条变更共享同一个 CSCN。

我的做法是把 CSCN 转成定长十六进制字符串存在 Flink 的 state 里，同时把(CSCN, 表名, 主键)三元组写进消息，这样下游既能做全局去重，也能做单行去重。

4.2 全量与增量怎么衔接：快照起点必须在加锁之前记

全量加增量是 CDC 最容易出错的地方，出错的表现往往是「快照期间变更的那几行，下游要么少一条，要么多一条」。根因在于快照读和日志读是两条独立的时间线，中间有个窗口期。

正确的顺序只有一种：先记下当前 CSCN 作为快照起点，再做全量读，读完从那个 CSCN 开始回放日志。这样窗口期内的变更会同时出现在快照结果和日志流里，靠主键 upsert 天然去重；反过来，如果先做快照再记位点，位点之前的那段变更就永久丢失了。

加锁与否是第二个选择。LOCK TABLE ... IN EXCLUSIVE MODE能保证快照期间没有写入，一致性最好，但大表上锁几分钟，业务方会来找你。我一般用分片快照：按主键范围把表切成若干片，每片单独记快照起点，片与片之间不加全局锁。代价是全局一致性从「严格」降到「片内严格」，能不能接受取决于业务，财务报表类的表我建议还是老实加锁，跑在凌晨。

大表分片还有个细节，分片边界最好用主键的数值范围而不是MOD取模，前者能走索引范围扫描，后者每片都要全表扫一遍。十万行以下的表感知不明显，上千万行的时候差距是分钟级的。

4.3 checkpoint 与位点提交：别自己往数据库写 offset

这条是血泪经验。早期我图省事，在解析循环里每处理完一批就把位点更新到一张 offset 表里，觉得这样重启后能精确续上。结果遇到一次作业被 kill 之后重启，offset 表里记的位置比实际发出去的数据要靠后，中间那段数据凭空消失，对账对了三天才定位到。

Flink 的容错语义是「checkpoint 完成才算数」，位点必须和 checkpoint 绑定。放在 state 里，由 Flink 在快照成功时统一持久化，作业恢复时从 state 读回来。自己在外面维护一套 offset，等于绕开了这套机制，两套记录必然会不一致。

还有一个更隐蔽的点：数据的发出和位点的更新必须在同一个原子区间内。如果先collect再更新位点，中间被 checkpoint 切开，恢复后这批数据会重发；如果先更新位点再collect，恢复后这批数据会丢。Flink 给的解法是ctx.getCheckpointLock()，把两件事锁在同一个临界区里。

4.4 Source 骨架：把上面三件事落到代码

下面这个骨架是简化版，去掉了连接池、指标上报和 DDL 处理，保留的是位点管理的主干，可以直接照着改。

public class DmLogSource extends RichSourceFunction<RowData> implements CheckpointedFunction { private static final int BATCH_SIZE = 1000; // 每批拉多少条日志记录 private static final long POLL_INTERVAL_MS = 500; // 空批次时的轮询间隔 private volatile boolean running = true; private transient Connection conn; private transient ListState<String> lsnState; private String currentLsn = null; @Override public void open(Configuration cfg) throws Exception { conn = DriverManager.getConnection(url, user, pwd); conn.setAutoCommit(false); // 解析会话不要自动提交 } @Override public void run(SourceContext<RowData> ctx) throws Exception { while (running) { List<Row> rows = fetchLogmnrRows(conn, currentLsn, BATCH_SIZE); if (rows.isEmpty()) { Thread.sleep(POLL_INTERVAL_MS); // 没有新日志就让 DB 喘口气 continue; } for (Row r : rows) { // 发出数据与推进位点必须在同一个临界区，否则恢复时会重或丢 synchronized (ctx.getCheckpointLock()) { ctx.collect(toRowData(r)); currentLsn = r.cscnHex(); } } } } @Override public void snapshotState(FunctionSnapshotContext c) throws Exception { lsnState.clear(); lsnState.add(currentLsn); // checkpoint 成功时由 Flink 统一持久化 } @Override public void initializeState(FunctionInitializationContext c) throws Exception { lsnState = c.getOperatorStateStore() .getListState(new ListStateDescriptor<>("lsn", String.class)); for (String s : lsnState.get()) { currentLsn = s; // 恢复时从上次快照的位点续读 } } @Override public void cancel() { running = false; } }

BATCH_SIZE和POLL_INTERVAL_MS是延迟与数据库压力的两个旋钮。1000 条一批、500 毫秒轮询，在每秒几百条变更的量级上大概对应 1 秒内的端到端延迟；把批次调大到 10000 能提升吞吐，但单批解析耗时变长，延迟反而上升，而且一旦失败重来的范围也更大。

fetchLogmnrRows里要处理两种边界：一是归档文件切换的瞬间，当前打开的文件读完了新的还没加进去，这时应该返回空列表而不是报错；二是起始位点对应的归档已经被清理，这种情况必须显式抛异常并告警，静默跳过等于丢数据。第二种情况我在测试环境故意造过一次，就是想看看代码会不会沉默地过去，结果它真的过去了，后来加了强制校验。

5. 避坑：五条我踩过的真实记录

这一章每条都是「现象 → 原因 → 解决」的结构，按我遇到的时间顺序排。有些坑一次就记住，有些我栽过两次。

5.1 归档目录写满，数据库直接挂起

现象：同步端突然报读不到日志文件，同时业务侧大面积超时，数据库连接全部卡住。

原因：归档目录所在磁盘写满，ARCH_SPACE_LIMIT设的是 0，也就是不限制，数据库一直往里面写，直到文件系统没有空间。数据库写归档失败会进入挂起状态，所有 DML 都被阻塞。这是最严重的一类故障，因为它影响的是主业务而不只是同步链路。

解决：三件事一起做。第一，ARCH_SPACE_LIMIT必须设一个明确的上限，留出至少 20% 的磁盘余量。第二，配一个清理任务，但清理条件不能只看文件修改时间，必须先去查解析端已经消费到哪个位点，确认归档文件已经被覆盖才删。第三，加磁盘水位告警，80% 就开始关注，90% 必须有人处理。

注意：清理脚本里千万不要写find /arch -mtime +3 -delete这种裸删除。归档保留天数必须大于你最长的一次故障恢复时间，否则一次长时间的作业中断之后，恢复所需的日志已经被自己的定时任务删干净了。

5.2 中文变成问号，en 和 cn 混在一张表里

现象：下游宽表里客户名称、地址这类字段全是?，但金额、编号这些数字字段完全正常，所以一开始没人发现。

原因：源库字符集和解析端的解码方式不一致。数字是 ASCII，怎么解都对；中文在解码错的情况下就变成占位符。这个问题的隐蔽之处在于它不会报错，数据安安静静地写进去了，等业务方查报表的时候才发现。

解决：全链路统一编码，从 JDBC 连接参数、解析端字符串处理、到 Kafka 的序列化器，全部显式指定 UTF-8，不依赖任何默认值。上线前拿一条含中文、含生僻字、含 emoji 的记录做端到端验证，光测中文汉字不够，有些字符集的兼容范围比想象中窄。

5.3 大字段解析出来是 null 或被截断

现象：商品描述、合同正文这类长文本字段，同步到下游全是 null，或者只剩前几十个字符。

原因：日志解析对 CLOB、BLOB、TEXT 这类大字段通常只记录定位信息，不记录完整内容，这是出于日志体积的

本文还有配套的精品资源，点击获取