百万级 JSON 数据同步优化：游标分页、分批聚合与断点续传

从 OOM 到 1.5 小时跑完：一次百万级数据同步任务的优化实践

最近做过一个百万级数据同步任务，业务逻辑本身并不复杂：从埋点表中读取用户设备里的 App 列表，解析 JSON，统计每个 App 出现的次数，最后写入一张汇总表。

真正麻烦的地方不是 SQL 怎么写，而是数据量上来以后，内存、分页性能、失败恢复和重复统计这些问题会一起出现。

这个任务前后经历了三个版本：

全量加载 ↓ Offset 分页 ↓ ID 游标分页 + 分批聚合 + checkpoint

第一版因为内存占用过高出现 OOM；第二版虽然解决了全量加载问题，但深分页越来越慢；最终改成 ID 游标分页后，任务可以稳定跑完，实际耗时约 1.5 小时。

这篇文章记录一下整个优化过程，以及其中几个比较容易踩坑的地方。

一、业务背景

源表里有一类 App 收集事件，type = 5。其中event字段是 JSON，保存着用户设备中安装的 App 列表。

简化后的数据大致如下：

{ "apps": [ {"name": "WhatsApp"}, {"name": "Facebook"}, {"name": "TikTok"} ] }

现在需要新增一张统计表：

id app_name app_count

用于统计每个 App 在所有用户设备中出现的总次数。

例如：

WhatsApp 120000 Facebook 98000 TikTok 76000

如果数据量不大，直接查询、解析、聚合就可以完成。但源表到了百万级之后，原来的实现很快就暴露出了问题。

二、第一版：全量加载导致 OOM

最开始的实现比较直接：

一次性查询全部 type = 5 的数据 ↓ 加载到 JVM 内存 ↓ 逐条解析 event JSON ↓ 使用 Map 统计 App 次数 ↓ 统一写入数据库

这种写法代码简单，但对内存非常不友好。

源表数据本身已经比较大，event又是 JSON 字符串。解析过程中还会创建：

• 查询结果对象；

• JSON 节点对象；

• App 列表；

• 临时字符串；

• 中间集合；

• 全局统计 Map。

也就是说，一条数据进入 JVM 后，占用的内存并不只是数据库中的原始字段大小。

随着数据不断加载，堆内存压力越来越大，GC 开始频繁，最终任务还没有处理完就出现了 OOM。

这个版本让我明确了一点：

百万级离线任务不能依赖一次性全量加载，必须从数据源开始分批处理。

三、第二版：Offset 分页越往后越慢

为了解决 OOM，第二版改成了分页查询：

SELECT id, event FROM source_table WHERE type = '5' ORDER BY id ASC LIMIT #{offset}, #{batchSize};

每次只查一批数据，处理完成后继续下一页。

这个版本确实解决了全量加载的问题，JVM 不再同时持有全部源数据。但任务运行一段时间后，又出现了新的问题：越往后查询越慢。

例如：

LIMIT 0, 1000; LIMIT 100000, 1000; LIMIT 1000000, 1000;

虽然每次最终只返回 1000 条数据，但 MySQL 仍然需要扫描并跳过 offset 前面的记录。

Offset 越大，需要跳过的数据越多。任务跑到后半段时，大量时间消耗在“找到这一页”上，而不是业务处理上。

根据已经完成的批次和剩余批次估算，这个版本完整跑完大约需要 8 小时。需要说明的是，8 小时并不是完整执行后的实测数据，因为任务耗时过长没有跑完，而是根据当时的处理进度估算出来的。

Offset 分页适合普通列表查询和浅分页，但不适合百万级全量扫描。

四、第三版：改成 ID 游标分页

最终将分页方式改成了 ID 游标分页：

SELECT id, event FROM source_table WHERE type = '5' AND id > #{lastId} AND id <= #{maxIdAtStart} ORDER BY id ASC LIMIT #{batchSize};

每批处理完成后，记录当前批次最大的 ID：

lastId = 当前批次最大 ID

下一批继续查询：

WHERE id > lastId

这种方式不需要 MySQL 每次重新跳过前面已经处理过的数据，而是直接基于索引继续向后扫描。

只要 ID 是唯一、稳定、可排序的字段，每批查询的性能就比较稳定。

这里有一个容易误解的地方：游标分页并不要求 ID 连续。

例如 ID 是：

1、2、5、9、20

查询条件：

WHERE id > 5 ORDER BY id ASC LIMIT 1000

仍然会正常读取 9、20，不会因为中间缺少 6、7、8 就漏数据。

五、为什么还要固定 maxIdAtStart

如果源表在同步期间仍然不断写入新数据，只使用：

WHERE id > lastId

任务就可能一直追着新数据向后跑。

在数据持续写入的情况下，全量任务的结束边界会变得模糊，极端情况下甚至一直无法完成。

因此，任务开始时先查询一次最大 ID：

SELECT MAX(id) FROM source_table WHERE type = '5';

将结果保存为：

maxIdAtStart

本次全量任务只处理：

id > lastId AND id <= maxIdAtStart

这样就相当于在任务启动时划定了一条边界：

本次全量任务 只处理启动前已经存在的数据

任务运行期间新增的数据，其 ID 会大于maxIdAtStart，不会混入本轮全量任务，后续再由增量任务处理。

这样全量和增量的边界就比较清楚：

第一次全量：0 < id <= maxIdAtStart 下一次增量： lastSyncedMaxId < id <= newMaxId

六、分批处理时，为什么不能维护一个全局 Map

分页只是控制了单次查询的数据量，但如果把每个批次的统计结果一直累加到一个全局 Map 中，内存占用还是会随着任务运行不断增长。

例如：

Map<String, Long> totalCountMap = new HashMap<>();

如果 App 名称种类多、数据质量不稳定，或者 JSON 中存在大量不同字符串，这个 Map 可能越来越大，再次产生内存压力。

最终采用的是批次级局部聚合：

查询一批数据 ↓ 当前批次内部解析 JSON ↓ 使用局部 Map 统计 appName -> count ↓ 批量写入统计表 ↓ 释放当前批次对象 ↓ 处理下一批

示意代码：

Map<String, Long> batchCountMap = new HashMap<>(); for (SourceRecord record : records) { List<String> appNames = parseAppNames(record.getEvent()); for (String appName : appNames) { batchCountMap.merge(appName, 1L, Long::sum); } } batchUpsert(batchCountMap);

一批处理完成后，records、JSON 临时对象和batchCountMap都可以被回收，内存不会随着任务进度持续上涨。

七、统计表如何处理跨批次累加

同一个 App 会出现在不同批次中，因此目标表需要支持累加。

先对app_name建立唯一索引：

UNIQUE KEY uk_app_name(app_name)

写入时使用 upsert：

INSERT INTO app_stat(app_name, app_count) VALUES (?, ?) ON DUPLICATE KEY UPDATE app_count = app_count + VALUES(app_count);

例如第一批统计出：

WhatsApp -> 500

第二批又统计出：

WhatsApp -> 300

最终表中会变成：

WhatsApp -> 800

这样每个批次只需要提交自己的增量，不需要在 JVM 中保存全量统计结果。

不过，这种count = count + delta并不是天然幂等的。

如果同一批数据被重复处理两次，统计值也会被累加两次。所以必须进一步解决 checkpoint 和批次事务问题。

八、最关键的事务边界：upsert 和 checkpoint 必须一起提交

为了支持任务中断后恢复，需要建立同步任务表，记录当前进度：

task_id task_type max_id_at_start last_id status start_time end_time error_msg

假设当前批次处理的是：

id 1001 到 2000

正确的执行顺序应该是：

1. 查询本批数据 2. 解析 JSON 3. 生成本批统计增量 4. 批量 upsert 到 app_stat 5. 更新 sync_task.last_id = 2000 6. 提交事务

其中：

统计结果 upsert + checkpoint 更新

必须放在同一个事务里。

否则会出现两种严重的不一致。

情况一：统计成功，checkpoint 没更新

如果 count 已经累加，但服务在更新lastId前宕机，任务重启后仍会从旧的lastId开始。

这一批数据会再次执行，最终导致重复统计。

情况二：checkpoint 更新了，统计没有成功

如果lastId已经推进到 2000，但统计结果写入失败，任务重启后会从 2000 之后继续。

那么 1001 到 2000 这一批就会被永久跳过，造成漏统计。

因此必须保证：

upsert 成功、lastId 更新成功：一起提交 任何一步失败：一起回滚

这个事务边界比使用哪种分页方式更重要。分页解决的是性能问题，事务解决的是正确性问题。

九、任务失败后如何继续执行

任务失败后，不能从头重新跑。

启动时先读取任务表中的：

last_id max_id_at_start

然后继续执行：

SELECT id, event FROM source_table WHERE type = '5' AND id > #{lastId} AND id <= #{maxIdAtStart} ORDER BY id ASC LIMIT #{batchSize};

由于lastId只会在当前批次完整提交后更新，因此：

• 已成功提交的批次不会重复处理；

• 未成功提交的批次会重新执行；

• 任务可以从最近一次成功位置继续。

这种机制本质上就是一个简单的 checkpoint。

十、遇到坏 JSON，不能让整个任务停下来

埋点数据很难保证完全干净。

可能存在：

• JSON 格式错误；

• 字段缺失；

• App 列表类型不正确；

• 单条 event 内容异常；

• 历史版本数据结构不一致。

如果任何一条坏数据都让整批回滚，百万级任务可能永远卡在同一条记录上。

但也不能简单 catch 后忽略，否则问题数据就彻底丢失了。

最终采用的是：

正常数据继续统计 异常数据写入异常表 主任务继续推进 后续通过补偿任务重新处理

异常表可以保存：

source_id task_id 原始 event error_msg status retry_count created_at updated_at

其中source_id建唯一索引，防止同一条异常记录被重复插入。

批次处理过程变成：

正常记录 -> 局部 Map 聚合 异常记录 -> errorList 最后： 批量 upsert 正常统计 批量写入异常表 更新 checkpoint 同一个事务提交

checkpoint 可以继续推进，但前提是异常记录已经可靠落表。

这样一条坏数据不会阻塞整个任务，同时也保留了后续修复的机会。

十一、补偿任务也要防止重复执行

补偿任务从异常表中扫描：

INIT RETRY

状态的数据，重新尝试解析和统计。

但补偿逻辑不能直接读取后就累加 count，否则多个补偿任务可能同时处理同一条记录，导致重复统计。

可以先通过条件更新抢占处理权：

UPDATE app_sync_error SET status = 'PROCESSING' WHERE source_id = ? AND status IN ('INIT', 'RETRY');

如果影响行数为 1，说明当前任务获得了处理权。

如果影响行数为 0，说明这条数据可能已经被其他任务处理，或者状态已经变化。

补偿成功后，需要在同一个事务里完成：

统计值累加 + 异常记录状态更新为 RESOLVED

如果只累加 count，没有更新状态，任务下次还会再次补偿；如果先更新状态，后续统计失败，又会造成漏数据。

对于多次补偿仍然失败的数据，可以转为：

FAILED

再通过告警或人工处理。

十二、如何防止两个同步任务同时运行

即使 checkpoint 和 upsert 放在同一个事务里，也不能完全防止人为重复启动任务。

例如两个全量任务同时从lastId = 0开始，它们可能读取到相同的数据，分别进行累加。

因此还需要任务互斥机制，例如：

任务状态检查 数据库唯一约束 Redis 分布式锁 调度平台禁止并发

最简单的方式是保证同一个任务类型，在同一时间只能存在一个RUNNING任务。

需要注意的是：

单任务串行 + upsert 与 checkpoint 同事务

能够覆盖正常的失败重试，但并不等于绝对强幂等。

如果业务对统计准确性、审计和回滚要求更高，可以进一步引入批次增量表。

十三、什么时候需要 batch delta 表

当前方案是直接把每个批次的统计增量累加到正式表。

如果需要更强的可追踪性，可以增加批次增量表：

task_id batch_no app_name delta_count

每个批次先把增量写入 delta 表，再根据task_id + batch_no保证批次唯一。

这样可以支持：

• 判断某个批次是否已经执行；

• 重新汇总统计结果；

• 回滚某个批次；

• 审计统计值来源；

• 防止重复累加。

但它也会增加数据量和系统复杂度。

对于普通风控统计任务，如果单任务执行、checkpoint 和异常补偿已经能满足要求，不一定需要一开始就上复杂方案。

技术设计还是要结合业务风险，而不是越复杂越好。

十四、线程池不是这次优化的核心

这个任务也可以通过线程池并行解析 JSON，或者并行写入数据库。

但线程池并不是解决 OOM 和深分页问题的核心手段。

盲目并发反而可能带来：

• 多批次同时驻留内存；

• 任务队列堆积；

• checkpoint 提交乱序；

• 数据库并发写入冲突；

• 下游压力过大；

• 失败恢复更加复杂。

因此更稳妥的顺序是：

先通过游标分页稳定查询 ↓ 通过分批处理控制内存 ↓ 通过事务保证进度正确 ↓ 确认真正瓶颈 ↓ 再决定是否并行

如果后续发现主要耗时在 JSON 解析，可以使用有界线程池处理当前批次，但必须限制队列长度，并确保批次完成后再推进 checkpoint。

不能为了提升吞吐，把任务进度的正确性打乱。

十五、索引也要配合分页方式调整

当前核心查询条件是：

WHERE type = '5' AND id > #{lastId} ORDER BY id ASC

可以考虑建立联合索引：

(type, id)

这样 MySQL 可以先根据type定位目标数据，再基于id做范围扫描。

是否真正使用该索引，还需要结合：

EXPLAIN

确认执行计划，不能只看 SQL 形式就默认索引一定有效。

十六、如果不能按 ID 同步怎么办

ID 游标分页要求游标字段稳定、唯一、可排序。

如果业务只能按create_time同步，就不能简单写：

WHERE create_time > #{lastCreateTime}

因为多条记录可能拥有相同的创建时间。

假设这一批只处理了某个时间点的一部分数据，就直接推进lastCreateTime，剩余同一时间的数据可能被跳过。

更稳妥的是使用复合游标：

WHERE ( create_time > #{lastCreateTime} OR ( create_time = #{lastCreateTime} AND id > #{lastId} ) ) ORDER BY create_time ASC, id ASC LIMIT #{batchSize};

也就是：

create_time + id

共同确定唯一排序位置。

游标分页的重点不是一定使用 ID，而是找到一组稳定且唯一的排序字段。

十七、最终效果

三个版本的变化可以总结为：

第一版：全量加载 结果：任务后期 OOM，无法完整执行 第二版：Offset 分页 结果：避免一次性全量加载，但深分页越来越慢 完整耗时约 8 小时为进度估算，并非完整实测 第三版：ID 游标分页 + 分批聚合 结果：任务可以完整执行，实际耗时约 1.5 小时 同时解决 OOM 问题

性能提升并不是来自某一个单独技巧，而是几项改造共同作用：

• 游标分页避免深分页扫描；

• 分批查询控制单批数据规模；

• 局部 Map 控制 JVM 内存；

• 批量 upsert 减少数据库交互；

• 合理索引提高范围扫描效率；

• checkpoint 避免失败后从头执行。

十八、总结

这次优化表面上是把 Offset 分页改成了 ID 游标分页，但真正需要解决的其实有四类问题：

查询性能 内存控制 失败恢复 统计正确性

最终形成的链路是：

任务开始时固定 maxIdAtStart ↓ 按照 id > lastId 分批扫描 ↓ 当前批次局部解析和聚合 ↓ 正常数据批量 upsert ↓ 异常数据可靠落表 ↓ 更新 checkpoint ↓ 同一个事务提交 ↓ 失败后从 lastId 继续

对于百万级离线任务来说，真正重要的不是把代码并发化，而是让每一批数据都满足：

查得稳定 内存可控 进度可恢复 结果不重不漏

只有这些基础边界确定之后，线程池、消息队列、CDC 等进一步优化才有意义。