百万级 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};每批处理完成后记录当前批次最大的 IDlastId 当前批次最大 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任务就可能一直追着新数据向后跑。在数据持续写入的情况下全量任务的结束边界会变得模糊极端情况下甚至一直无法完成。因此任务开始时先查询一次最大 IDSELECT MAX(id) FROM source_table WHERE type 5;将结果保存为maxIdAtStart本次全量任务只处理id lastId AND id maxIdAtStart这样就相当于在任务启动时划定了一条边界本次全量任务 只处理启动前已经存在的数据任务运行期间新增的数据其 ID 会大于maxIdAtStart不会混入本轮全量任务后续再由增量任务处理。这样全量和增量的边界就比较清楚第一次全量0 id maxIdAtStart 下一次增量 lastSyncedMaxId id newMaxId六、分批处理时为什么不能维护一个全局 Map分页只是控制了单次查询的数据量但如果把每个批次的统计结果一直累加到一个全局 Map 中内存占用还是会随着任务运行不断增长。例如MapString, Long totalCountMap new HashMap();如果 App 名称种类多、数据质量不稳定或者 JSON 中存在大量不同字符串这个 Map 可能越来越大再次产生内存压力。最终采用的是批次级局部聚合查询一批数据 ↓ 当前批次内部解析 JSON ↓ 使用局部 Map 统计 appName - count ↓ 批量写入统计表 ↓ 释放当前批次对象 ↓ 处理下一批示意代码MapString, Long batchCountMap new HashMap(); for (SourceRecord record : records) { ListString 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)写入时使用 upsertINSERT 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 等进一步优化才有意义。