DataX 增量同步 MySQL 到 ES 的工程实践与踩坑指南

做数据同步这活儿，写脚本不难，难的是把“同步”这件事长期维持住。我之前接了一个订单查询的需求，MySQL 里一张订单表已经上亿行，前端搜索要查 Elasticsearch，一开始图省事，每天凌晨全量跑一遍 DataX 任务。结果每天全量耗时越来越长，数据库压力也越来越大，后来把方案改成 DataX 增量同步，才真正把这件事理顺。这篇文章就围绕“MySQL 同步数据到 ES”这个场景，把从配置到调度、从踩坑到稳定的完整经验写出来，非常适合正在做订单、交易、日志类数据同步的研发，尤其是准备用 DataX 做定时增量、又不想引入太重实时链路的团队参考。

1. 为什么最后选了 DataX 来做 MySQL 到 Elasticsearch 的增量同步

1.1 拿到需求后我把几个同步方案比了一遍

当时业务方只提了一个硬需求：订单数据要进 Elasticsearch，支持按订单号、买家昵称、商品名称搜索，延迟能接受一两分钟。一听“延迟一两分钟能接受”，我心里就把实时链路排除了大半。常见的做法无非三种：自己写 Java 定时任务、基于 binlog 做实时同步、用 DataX 这类离线同步框架。

自研看起来灵活，但要处理分页、批量写 ES、异常重试、增量断点，写着写着就变成一个小项目。实时同步是强，但要维护采集进程、消费位点和消息队列，对一个内部搜索场景来说有点重。DataX 的好处在于 MySQL Reader 和 ES Writer 都是现成的，我只需要关心“查哪些数据”和“怎么把任务调度起来”。所以最后选了 DataX。

1.2 DataX 的“增量同步”到底是什么

DataX 本身是一个离线同步执行框架，一个任务启动后，把源端数据读出来，经过通道并发处理，再写到目标端。它没有内置“持续侦测变更”的能力。所谓增量同步，本质上是利用 MySQL 查询条件把每次要读的数据范围卡小，比如id > 上次的最大id，或者update_time >= 上次同步时间。

DataX 只负责在给定查询范围内搬运数据；游标记录在哪里、每次传什么条件给 DataX，这些都要在任务外层做。理解这一点很重要，因为很多人以为缺一个配置就能自动增量，实际上增量是一种工程约定，不是框架内置开关。

1.3 增量同步适合什么样的业务

• MySQL 数据量持续增长，全量同步已经跑不动；
• 业务能接受分钟级延迟，不需要秒级实时；
• 表里存在自增主键或者“最后更新时间”字段；
• 删除操作不多，或者表里已经有软删除标记；
• 需要支持大表批量同步，不想为一个小需求专门搭一套实时计算平台。

反过来，如果业务要求秒级更新、大量物理删除、需要事务一致性，那 DataX 增量同步做起来会很别扭，这种情况不如直接上 binlog 类的实时采集方案。我在后面还会专门讲这个边界。

2. 增量同步方案设计：主键游标、时间戳与删除标记

2.1 主键游标和时间游标怎么选

增量方案设计，第一个问题就是“靠什么判断哪些数据是新的”。最简单的是自增主键游标。每次记录同步到的最大 id，下一次任务只要跑where id > ${lastMaxId}。这个方案性能最好，因为主键索引定位很快，但如果业务会把历史订单的某些字段更新，比如修改收货地址、退款状态，这些行只靠 id 游标是扫不到的，因为它们的 id 并没有变。

第二种是时间游标。用update_time字段筛选最近变化的数据，条件写成update_time >= '上次同步时间' and update_time < '本次同步时间'。它能捕获到新增和更新两类动作，前提是 update_time 要加索引，否则大表全表扫描会把你数据库拖垮。很多订单表会把 create_time 和 update_time 都加上，增量同步应该用 update_time，而不是 create_time，否则只有新增能被捞到，更新还是漏。

实际业务里，我一般这样分：

如果有些表实在没有 update_time 字段，只能退而求其次用主键游标，但要在需求文档里和业务方确认“已变更的历史数据不需要同步”或“可以接受定期全量补偿”。

2.2 游标状态放在哪：状态表比本地文件可靠

游标值不能只放在脚本变量里，否则任务一重启就丢了。常见做法是放在一个状态表里，比如：

CREATE TABLE sync_state ( task_name VARCHAR(128) PRIMARY KEY, last_value VARCHAR(64) NOT NULL, last_sync_time DATETIME NOT NULL, last_status VARCHAR(16) DEFAULT NULL );

每次增量任务启动前，先查这张表拿到 last_value；任务成功结束后，再把 last_value 更新成本次任务的时间或最大 id。用表存游标，最大的好处是多个调度节点、多个脚本可以共享同一个状态，排查问题也方便。本地文件不是不能用，但 DataX 常常是在脚本里被拉起来，文件路径一旦写死，任务迁移或者多机并发就容易出乱子。

2.3 更新和删除怎么处理：文档 ID 与软删除

把 MySQL 数据写到 ES 时，最要紧的一件事就是把业务主键映射成 ES 文档的_id。比如订单表的主键是 id，那么 ES 里每个文档_id也应该是 id。这样同一行数据不管被同步多少次，写进去都是同一个文档，天然覆盖更新。

如果_id不设，ES 会自动生成随机 ID，那么每次同步都会产生新文档，索引里全是重复数据，这也是增量同步最容易踩的坑。

删除是个更麻烦的问题。DataX 的 MySQL Reader 只负责读出符合条件的数据，不会主动去删除 ES 里已经存在的文档。如果 MySQL 里某一行被物理删除，DataX 不会知道，ES 里那个文档就一直残留。我的处理办法有几个：一是业务上加软删除，也就是is_deleted字段，查询时把删除标记带上，ES 侧也把这个字段写进文档，前端搜索时过滤掉；二是定期跑一个全量校准任务，重建索引，然后用索引别名切换；三是如果删除量很大且需要即时生效，就不要用 DataX 做增量，直接考虑实时同步。后面第 5 章我还会再展开校准的做法。

3. DataX 从 MySQL 到 ES 的 job 配置实战

3.1 准备阶段：先把插件和环境检查清楚

DataX 解压之后，先确认plugin/writer/elasticsearchwriter这个目录存在。有些精简版发行包只带常用 Writer，如果缺 ES 插件，后面的 job 配置写得再对也起不来。检查完之后，最好先用一个自建小索引做通读测试，确认 MySQL 驱动版本和 ES 的版本兼容。

ES 开启了安全认证的话，Writer 参数里要按对应插件的规范填写认证信息；不同版本的 elasticsearchwriter 对安全参数的字段叫法可能不一样，一定要以你手里那版插件自带的 json 模板为准。另外，我建议给 MySQL 单独建一个同步账号，只给查询权限。DataX 任务常常被调度系统拉起，账号权限越小越安全，也避免业务库误操作。

3.2 MySQL Reader 怎么配置才不容易翻车

常见的 mysqlreader 配置结构是这样的：

"reader": { "name": "mysqlreader", "parameter": { "username": "sync_user", "password": "sync_pass", "column": [ "id", "order_no", "buyer_name", "pay_amount", "order_status", "update_time", "is_deleted" ], "splitPk": "id", "where": "update_time >= '${lastTime}' AND update_time < '${currentTime}'", "connection": [ { "jdbcUrl": ["jdbc:mysql://192.168.1.10:3306/mall_order?useSSL=false&useUnicode=true&characterEncoding=utf8&serverTimezone=Asia/Shanghai"], "table": ["t_order"] } ] } }

这里最关键的几个地方：

• column要和 ES mapping 里需要的字段对齐，不要一股脑select *。数据量大时，多一个没用的长字段，ES network 压力和内存占用都会增加。
• where是增量核心。${lastTime}和${currentTime}会在 DataX 启动时通过-p参数替换。条件必须是半开区间，也就是说左边包含、右边不包含，左边是上次成功跑完的时间，右边是本次任务启动时间，这样不会因为边界重复而导致同一行被反复读出来。
• splitPk填 id，DataX 才能把查询拆分成多个范围并发读取。如果数据量小，不填也可以，但填了能明显提升大范围查询速度。
• jdbcUrl里我习惯带上characterEncoding=utf8和serverTimezone=Asia/Shanghai，不然中文和时区问题会以非常诡异的方式冒出来。

有些数据源的版本不允许在 Reader 里直接写where，你可以在 connection 里写querySql把完整select语句直接放进去。两种方式能跑通一种就行，我更喜欢where，因为它和splitPk配合起来比较自然。

3.3 Elasticsearch Writer 配置要点

再来看 Writer：

"writer": { "name": "elasticsearchwriter", "parameter": { "endpoint": "http://192.168.1.20:9200", "accessId": "", "accessKey": "", "index": "mall_order_search", "type": "_doc", "batchSize": 1000, "primaryKeyInfo": { "name": "id", "type": "id" } } }

primaryKeyInfo在不同版本的插件里写法略有出入，有的叫primaryKeyInfo，有的用别的参数结构，核心目的都是把 MySQL 主键映射成 ES 文档_id。这一步不能省。如果配置里没有类似字段，请去翻你插件目录下自带的模板，照着补上。

batchSize控制一次 bulk 请求塞多少条，默认值通常还能用，但具体多少要看 ES 的 bulk 承受能力。数据量大时，我建议从 1000 左右开始压测，而不是一上来就写几千。type在 ES 6 之后已经弱化，ES 7/8 通常写_doc，这个跟着你的 ES 版本走。

另外要特别提醒：如果索引是 DataX 自动创建的，字段类型不一定符合你的搜索需求。比如中文全文检索，自动 mapping 很可能把文本字段建成keyword，到时候想全文搜索却只能精确匹配。所以正规做法是先在 ES 里把 index mapping 建好，再把 DataX 的字段对齐，而不要让同步工具替你决定模型。

3.4 一套可运行的增量 job 配置模板

把 Reader 和 Writer 放到一个完整 job 里，再加一个简单的 setting：

{ "job": { "setting": { "speed": { "channel": 2 }, "errorLimit": { "record": 0, "percentage": 0.02 } }, "content": [ { "reader": { "name": "mysqlreader", "parameter": { "username": "sync_user", "password": "sync_pass", "column": ["id", "order_no", "buyer_name", "pay_amount", "order_status", "update_time", "is_deleted"], "splitPk": "id", "where": "update_time >= '${lastTime}' AND update_time < '${currentTime}'", "connection": [ { "jdbcUrl": ["jdbc:mysql://192.168.1.10:3306/mall_order?useSSL=false&useUnicode=true&characterEncoding=utf8&serverTimezone=Asia/Shanghai"], "table": ["t_order"] } ] } }, "writer": { "name": "elasticsearchwriter", "parameter": { "endpoint": "http://192.168.1.20:9200", "index": "mall_order_search", "type": "_doc", "batchSize": 1000, "primaryKeyInfo": { "name": "id", "type": "id" } } } } ] } }

errorLimit.record设置为 0，代表一条脏数据都不允许，这对增量同步来说更严格、不容易把坏数据悄悄写进 ES。percentage我一般配置成 0.02，当数据量特别大时，允许极少量脏数据不打断整个任务，但这个值不能拍脑袋调大，否则问题数据会被掩盖。channel不只是越大越好，它和batchSize、JVM 内存是联动关系，盲目调大反而容易把 ES 的 bulk 接口打挂或让 DataX 自己 OOM。

4. 增量调度与游标管理：让同步任务自动、可靠地跑

4.1 一个调度脚本需要做什么

job.json 只是中间环节，真正让增量跑起来的是外层脚本。逻辑大概是这样：

#!/bin/bash TASK_NAME="mall_order_to_es" LAST_TIME=$(mysql -N -h 192.168.1.10 -u sync_user -p"sync_pass" mall_order \ -e "SELECT last_value FROM sync_state WHERE task_name='${TASK_NAME}';") if [ -z "$LAST_TIME" ]; then LAST_TIME="1970-01-01 00:00:00" fi CURRENT_TIME=$(date '+%Y-%m-%d %H:%M:%S') python /opt/datax/bin/datax.py -p "-DlastTime=${LAST_TIME} -DcurrentTime=${CURRENT_TIME}" \ /opt/datax/job/mall_order_to_es.json if [ $? -eq 0 ]; then mysql -h 192.168.1.10 -u sync_user -p"sync_pass" mall_order \ -e "UPDATE sync_state SET last_value='${CURRENT_TIME}', last_sync_time=NOW(), last_status='SUCCESS' WHERE task_name='${TASK_NAME}';" fi

这个脚本看起来简单，但有几件事不能省：

• 第一次跑的时候，sync_state 里没有记录，要给LAST_TIME一个初始值，我常用1970-01-01 00:00:00，这样第一次任务就相当于从全表开始拉。
• CURRENT_TIME取的是脚本启动时间，而不是 DataX 跑完的时间。取跑完时间反而有陷阱：任务跑 10 分钟，期间新产生的数据会被漏到下一次窗口之外，导致空档。
• 状态更新只放在 DataX 成功退出之后。如果任务失败，last_value保持不变，下次重跑还能从上次成功的点继续，这就叫断点续传，也是增量同步最核心的可靠性保证。

注意：如果状态值里有空格，-p "-DlastTime=... -DcurrentTime=..."这种传参在不同 DataX 版本里可能会有解析问题。我的规避方法是把 last_value 统一存成yyyyMMddHHmmss，job 里的 where 写成update_time >= STR_TO_DATE('${lastTime}', '%Y%m%d%H%i%s')，这样参数里没有空格，也照样能走 update_time 索引。这个细节是我实际踩过坑才记住的。

4.2 幂等和重复同步问题

增量任务失败后重跑，或者状态更新失败，都可能让同一个时间窗口的数据被反复读到。这里不用担心 ES 出现大量重复，因为_id等于业务主键，重复写同一个文档就是覆盖更新。但前提是primaryKeyInfo配好了。如果没配，反复跑就等于反复新增文档，索引里的重复数据会越来越多。

所以我在自己的任务里养成了一个习惯：检查增量任务是否幂等，不看脚本，只看_id有没有落对。ES 里执行一个按业务主键聚合的查询，能清楚地看到哪些文档 id 重复了。出现重复，不用慌，把索引删了重新跑一次全量任务，再从增量开始，比写一堆去重逻辑更干净。

4.3 调度频率、并发和任务重叠怎么处理

调度频率没有统一标准。我之前在订单搜索场景用的是每 5 分钟一次，update_time 上有索引，单次窗口查询基本在毫秒级；对数据实时性要求稍高的同事会把频率调到 1 分钟。调节奏时重点看两个指标：MySQL 慢查询量和 ES 的 bulk 耗时。如果窗口内数据量很大，单次任务超过调度周期，就一定会出现任务重叠。

同一张表的增量任务不建议并发跑。两个任务同时读同一窗口，ES 写入时会出现大量争抢，状态表的 last_value 也会被后结束的那个任务覆盖。稳妥的办法是在调度平台上给同一表任务加互斥锁，或者干脆用单机 crontab，保证同一时间只跑一个 DataX 进程。

5. 实操中踩过的坑：连接中断、mapping、时区与脏数据

5.1 MySQL 连接闪断和管道中断

增量任务跑在半路报Connection is closed、broken pipe这类错误，是最常见的问题。原因通常是单个 MySQL 连接被网络层或 MySQL 端断开，但 DataX 的 channel 还在等数据。

我处理这个问题的顺序是先检查任务是不是跑得太久，比如一张大表在没有合理增量条件下被当成全量扫；再去jdbcUrl里加 socket 超时参数，避免连接长时间空闲被服务端判定为死连接；最后看是不是channel开得太多，把数据库连接池打爆了。DataX 的日志里会记录具体哪一步失败，不要只看最后一行错误提示，往前翻几十行找真正断掉的位置。

5.2 JVM 内存和批量参数的调参经验

DataX 自带脚本里的 JVM 默认堆内存并不大，我接手过不少任务，数据量一上来直接 OOM。这时候不要急着加 channel，先看两处：datax.py里的-Xms和-Xmx，以及 Writer 的batchSize。

如果源表几千万行，堆内存不调到 2GB 以上，channel 再大也只是把内存耗得更快。批量大小也不是越大越高效，ES bulk 请求太大反而会触发拒绝响应，返回 429。我的经验是从channel=2、batchSize=1000开始，逐步增加，观察 DataX 的吞吐和 ES 的 CPU、bulk rejected 数，找到平衡点。

还要注意一个细节：如果字段里有很大的文本，比如商品详情、售后记录，单条记录本身就可能几十 KB，这时候 batchSize 一定要降下来，否则一次 bulk 数据量会很恐怖。批量大小按“条数”控制，不代表“字节数”可控。

5.3 ES mapping 不提前建，后面很被动

我在 3.3 里已经强调过 mapping，这里再说一个实操教训。自动创建的索引，经常会把所有数值字段推断成long或double，把订单号这种本应keyword的字段弄成text，结果前端做精确过滤和排序时全都失效。

提前建 mapping 的收益不只是性能，更是让你有机会把字段的可搜索语义定下来。比如订单号、状态码这些用于精确匹配和聚合的字段用keyword，搜索字段用text；金额字段用scaled_float或字符串，避免浮点精度问题；时间字段显式指定yyyy-MM-dd HH:mm:ss，就不会出现时区和格式解析的歧义。

如果 mapping 建的不对，DataX 不会帮你改，ES 在写入时会拒绝类型不兼容的数据，然后抛到脏数据统计里。所以同步之前先花半小时把 mapping 建好，能省后面好几个小时的排查时间。

5.4 日期时区问题

MySQL 的datetime和 ES 的date都带时区语义，DataX 做转换时如果 JDBC 连接没有指定serverTimezone，结果很可能和预期差 8 小时。同一个订单，在 MySQL 里是 12:00，到 ES 变成 20:00，搜索时间范围就乱了。

我在jdbcUrl里统一加了serverTimezone=Asia/Shanghai，并且让 ES mapping 里时间字段的 format 和 MySQL 输出格式一致，这个问题基本没有再犯过。如果你所在团队的服务部署在 UTC 环境，也要先明确业务统一用哪个时区展示，不要在同步链路里混合时区。

5.5 脏数据和数据校验怎么盯

DataX 任务跑完不代表数据就是对的。我还要做一个比较笨但有效的校验：每次增量跑完后，在 MySQL 里查窗口内数据的 count，在 ES 里查相同窗口的文档数，两边的量差在合理范围内才算成功。

如果差太多，优先看脏数据日志，DataX 会打印出具体字段；如果业务能容忍少量坏行，可以把errorLimit.percentage调高一点，但我倾向于保持严格，让脏数据立刻暴露在任务日志里，而不是藏到 ES 里再返工。

6. 这套方案的适用边界与我的使用体会

6.1 把增量模板沉淀成通用能力

用 DataX 做 MySQL 到 ES 增量同步，跑通一两个任务之后，剩下的工作其实是复制粘贴。真正降低维护成本的做法，是把状态表、调度脚本、job.json 模板整理成一套通用脚本，新增一张表只改表名、字段列表、索引名和主键映射。

我在实际项目中就是这么做的，同步任务从几十个慢慢变多之后，没有靠人工维护每个任务的差异，而是把差异性收敛到一张任务配置表里，让脚本去生成 job 文件。这样加新表就像加一行配置，而不是又写一个独立的 bash 任务。

6.2 什么情况下不要硬用 DataX

DataX 增量同步不是银弹。如果业务对延迟要求是秒级，或者表里的物理删除和频繁更新很多，DataX 的定时增量天然不匹配。这种场景应该考虑基于 binlog 的实时同步方案，把变更事件准确捕获到消息队列里，再消费进 ES。

DataX 的优势在批量、稳定、简单，适合那些能接受分钟级延迟、以新增为主、偶尔更新的表。同样的表结构，如果每天凌晨全量跑已经能接受，那也不一定非要切增量；一旦全量跑不动了，增量方案再上。

另外，DataX 增量同步对“删除”的支持是残缺的，这一点必须在方案阶段就跟业务方说清楚。不要等到 ES 里出现大量已删除数据才想起要处理。要么业务软删除，要么定期全量校准，要么换实时方案，三者必须选一个。

6.3 一点个人体会

最后分享一个我自己的体会：增量同步做得好不好，往往不看 DataX 写得多少花哨，而看状态管理和监控是不是到位。游标更新、失败重跑、幂等校验、脏数据日志，这些才是增量任务长期稳定的底盘。

把每次都跑得很顺当成目标，不如把“跑挂了也能安全重跑”当成目标。每次新接一个同步需求，我都会先问三件事：有没有可靠且递增的字段、能不能接受分钟级延迟、删除数据怎么处理。这三个问题想清楚了，DataX 增量同步的路子基本就稳了。