- 背景:本项目是从真实实践过的业务项目中拆分出来的,代码全部由本人独自开发
- 实践:进行过线上两次数据迁移和多次数据清理工作,数据量级(亿级别,理论无上限)
- 适用场景:【线上不停机】项目重构时的数据迁移,脏数据清理,更换数据库(eg,Qracle >> MySQL),数据修复等操作,且支持跨实例操作
- 性能:视具体数据库性能而定(只要数据库性能足够好,此项目的处理性能就足够高)
日志排查 Log:
处理的ID范围
- 数据迁移表示例
| 数据迁移顺序 | data:migrate:task | data:migrate:offset | .threadNum (默认) | .qps (默认) | .everyCommitCount (默认) |
|---|---|---|---|---|---|
| 1 | message.migrate | source.message#id in [15550, 17037] | 4 | 900 | 1500 |
| ... | ... | ... | ... | ... | ... |
- 数据清理表示例
| 数据清理顺序 | data:migrate:task | data:migrate:offset | .threadNum (默认) | .qps (默认) | .everyCommitCount (默认) |
|---|---|---|---|---|---|
| target.message【截止5月2日10点,2310087836条数据】 | message.clean | target.message#message_id in [16187, 17037] | 4 | 900 | 1500 |
| ... | ... | ... | ... | ... | ... |
操作步骤:
- 调用
updateValueByKey接口更新data:migrate:offset的值(迁移或清理数据时的段起始值,具体含义可以去看代码) - 调用
updateValueByKey接口更新data:migrate:task为具体的 task,开始进行数据迁移 or 数据清理任务 - 观察日志,如果发现处理的数据已经超过了想要处理的阈值,调用
updateValueByKey接口更新value为空字符串,来停止 task