档案数据互通全流程实操指南 跨系统无差错对接零门槛落地方法

一、前置环境准备

1.1 必备工具清单

以下工具均提供官方稳定版下载地址，可直接获取：

• JDK 11：https://adoptium.net/zh-CN/temurin/releases/?version=11
• Canal 1.1.6（数据采集中间件）：https://github.com/alibaba/canal/releases/download/canal-1.1.6/canal.deployer-1.1.6.tar.gz
• Flink 1.17（数据清洗计算引擎）：https://archive.apache.org/dist/flink/flink-1.17.0/flink-1.17.0-bin-scala_2.12.tgz

必须提前开放服务器对应端口：Canal的11111端口、MySQL的3306端口、Flink的8081端口，执行命令： firewall-cmd --add-port=11111/tcp --permanent && firewall-cmd --add-port=3306/tcp --permanent && firewall-cmd --add-port=8081/tcp --permanent && firewall-cmd --reload

1.2 源库基础配置

源端为MySQL的档案库必须开启Binlog，修改my.cnf（或my.ini）配置文件，完整内容如下：

```ini [mysqld] log-bin=mysql-bin binlog-format=ROW server_id=1 expire_logs_days=30 ```

修改后重启MySQL服务，执行命令验证开启状态：show variables like '%log_bin%';，log_bin值为ON即配置成功。

给Canal账号授权，执行SQL：GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON . TO 'canal'@'%' IDENTIFIED BY '自定义密码'; FLUSH PRIVILEGES;

二、核心对接流程实操

2.1 源端档案数据采集配置

解压Canal安装包后，修改conf/example/instance.properties配置文件，完整可复制内容如下：

```properties canal.instance.master.address=源库IP地址:3306 canal.instance.master.journal.name= canal.instance.master.position= canal.instance.master.timestamp= canal.instance.dbUsername=canal canal.instance.dbPassword=你设置的canal账号密码 canal.instance.connectionCharset = UTF-8 只同步档案相关表，可根据实际表名调整规则 canal.instance.filter.regex=.\\.archive_. canal.instance.filter.black.regex= ```

执行启动命令：sh bin/startup.sh，查看logs/example/example.log，出现“start successfully”即采集服务启动成功。

2.2 数据清洗转换规则配置

启动Flink服务后进入Flink SQL客户端，执行以下代码完成清洗规则配置，可直接复制使用：

```sql -- 关联Canal采集的源端档案表 CREATE TABLE archive_source ( id BIGINT, archive_no STRING, id_card STRING, name STRING, mobile STRING, create_time TIMESTAMP, update_time TIMESTAMP, PRIMARY KEY (id) NOT ENFORCED ) WITH ( 'connector' = 'canal', 'hostname' = 'Canal服务IP地址', 'port' = '11111', 'destination' = 'example', 'format' = 'canal-json' ); -- 清洗后的中间表 CREATE TABLE archive_clean ( id BIGINT, archive_no STRING, id_card STRING, name STRING, mobile STRING, create_time TIMESTAMP, update_time TIMESTAMP, PRIMARY KEY (id) NOT ENFORCED ) WITH ( 'connector' = 'print' ); -- 执行清洗逻辑 INSERT INTO archive_clean SELECT id, LPAD(TRIM(archive_no),15,'0') as archive_no, -- 统一档案编号为15位，不足补前导0 CONCAT(LEFT(id_card,6),'',RIGHT(id_card,4)) as id_card, -- 身份证脱敏 name, CONCAT(LEFT(mobile,3),'',RIGHT(mobile,4)) as mobile, -- 手机号脱敏 create_time, update_time FROM archive_source -- 过滤无效数据 WHERE archive_no IS NOT NULL AND id_card REGEXP '^[1-9]\\d{5}(18|19|20)\\d{2}((0[1-9])|(1[0-2]))(([0-2][1-9])|10|20|30|31)\\d{3}[0-9Xx]$' AND mobile REGEXP '^1[3-9]\\d{9}$'; ```

必须添加数据质量校验规则，引入Deequ 2.0.1依赖，执行校验逻辑：只有档案编号唯一性100%、身份证格式合规率≥99.99%的数据才允许流入目标端。

2.3 目标端数据写入配置

目标端为政务档案库时，先在目标库建表并添加唯一索引防止重复数据，执行SQL：ALTER TABLE archive_info ADD UNIQUE KEY uk_archive_no(archive_no);

在Flink SQL中添加目标表配置，开启幂等写入：

```sql CREATE TABLE archive_target ( id BIGINT, archive_no STRING, id_card STRING, name STRING, mobile STRING, create_time TIMESTAMP, update_time TIMESTAMP, PRIMARY KEY (archive_no) NOT ENFORCED ) WITH ( 'connector' = 'jdbc', 'url' = 'jdbc:mysql://目标库IP:3306/目标库名?useUnicode=true&characterEncoding=utf8&useSSL=false', 'username' = '目标库账号', 'password' = '目标库密码', 'table-name' = 'archive_info', 'sink.buffer-flush.max-rows' = '1000', -- 批量1000条提交 'sink.max-retries' = '3' -- 写入失败重试3次 ); -- 写入目标库 INSERT INTO archive_target SELECT FROM archive_clean; ```

三、异常排查与回滚方案

3.1 常见问题排查

• Canal采集中断：执行命令tail -f /usr/local/canal/logs/example/example.log排查，常见原因有Binlog过期、源库权限变更、网络不通
• 数据写入失败：查看Flink任务报错，常见原因有字段长度不匹配、唯一索引冲突、目标库权限不足
• 同步延迟过高：调整Flink并行度至和源库表分片数一致，默认并行度设置为4即可满足百万级档案同步需求

3.2 数据回滚操作

回滚前必须全量备份当前目标库数据，执行命令：mysqldump -u目标库账号 -p目标库密码 目标库名 > /data/backup/archive_backup_$(date +%Y%m%d).sql

按时间点回滚操作：mysqlbinlog --start-datetime="出错前的时间" --stop-datetime="出错后的时间" 源库Binlog文件路径 | mysql -u目标库账号 -p目标库密码 目标库名

四、常态化校验机制配置

4.1 每日数据对账配置

创建定时对账脚本archive_check.sh，设置crontab每天凌晨2点执行，可直接复制使用：

```shell !/bin/bash 统计前一天源端新增档案数 source_count=$(mysql -h源库IP -u源库账号 -p源库密码 -N -e "select count() from archive_info where create_time >= date_sub(curdate(),interval 1 day)") 统计前一天目标端新增档案数 target_count=$(mysql -h目标库IP -u目标库账号 -p目标库密码 -N -e "select count() from archive_info where create_time >= date_sub(curdate(),interval 1 day)") if [ $source_count -ne $target_count ] then echo "`date +%Y-%m-%d` 档案同步差异：源端新增$source_count 目标端新增$target_count" >> /var/log/archive_check.log 对接企业微信机器人告警，替换为自己的机器人key curl "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=你的机器人key" -H "Content-Type: application/json" -d '{"msgtype":"text","text":{"content":"档案同步异常：源端新增'$source_count' 目标端新增'$target_count'"}}' fi ```

4.2 告警规则配置

• 同步延迟≥10分钟触发告警
• 数据差异率≥0.01%触发告警
• Canal、Flink服务中断触发告警

按照以上流程操作即可完成跨系统档案数据互通，全流程无额外依赖，所有配置可直接复制落地。