給你10億數(shù)據(jù)，如何做遷移？
前言
某次金融系統(tǒng)遷移項目中，原計劃8小時完成的用戶數(shù)據(jù)同步遲遲未能完成。
24小時后監(jiān)控警報顯示：由于全表掃描SELECT * FROM users導(dǎo)致源庫CPU幾乎熔毀，業(yè)務(wù)系統(tǒng)被迫停機(jī)8小時。
這讓我深刻領(lǐng)悟到——10億條數(shù)據(jù)不能用蠻力搬運(yùn)，得用巧勁兒遞接！
今天這篇文章，跟大家一起聊聊10億條數(shù)據(jù)，如何做遷移，希望對你會有所幫助。
一、分而治之
若把數(shù)據(jù)遷移比作吃蛋糕，沒人能一口吞下整個十層蛋糕；
必須切成小塊細(xì)嚼慢咽。
避坑案例：線程池濫用引發(fā)的血案
某團(tuán)隊用100個線程并發(fā)插入新庫，結(jié)果目標(biāo)庫死鎖頻發(fā)。
最后發(fā)現(xiàn)是主鍵沖突導(dǎo)致——批處理必須兼顧順序和擾動。
分頁遷移模板代碼：
long maxId = 0;
int batchSize = 1000;
while (true) {
List<User> users = jdbcTemplate.query(
"SELECT * FROM users WHERE id > ? ORDER BY id LIMIT ?",
new BeanPropertyRowMapper<>(User.class),
maxId, batchSize
);
if (users.isEmpty()) {
break;
}
// 批量插入新庫（注意關(guān)閉自動提交）
jdbcTemplate.batchUpdate(
"INSERT INTO new_users VALUES (?,?,?)",
users.stream().map(u -> new Object[]{u.id, u.name, u.email}).collect(Collectors.toList())
);
maxId = users.get(users.size()-1).getId();
}避坑指南：
- 每批取遞增ID而不是OFFSET，避免越往后掃描越慢
- 批處理大小根據(jù)目標(biāo)庫寫入能力動態(tài)調(diào)整（500-5000條/批）
二、雙寫
經(jīng)典方案是停機(jī)遷移，但對10億數(shù)據(jù)來說停機(jī)成本難以承受，雙寫方案才是王道。
雙寫的三種段位：
- 青銅級：先停寫舊庫→導(dǎo)數(shù)據(jù)→開新庫 →風(fēng)險：停機(jī)時間不可控
- 黃金級：同步雙寫+全量遷移→差異對比→切流 →優(yōu)點(diǎn)：數(shù)據(jù)零丟失
- 王者級：逆向同步兜底（新庫→舊庫回寫），應(yīng)對切流后異常場景
當(dāng)然雙寫分為：
- 同步雙寫
- 異步雙寫
同步雙寫實(shí)時性更好，但性能較差。
異步雙寫實(shí)時性差，但性能更好。
我們這里考慮使用異步雙寫。
異步雙寫架構(gòu)如圖所示：
圖片
代碼實(shí)現(xiàn)核心邏輯：
- 開啟雙寫開關(guān)
@Transactional
public void createUser(User user) {
// 舊庫主寫
oldUserRepo.save(user);
// 異步寫新庫（允許延遲）
executor.submit(() -> {
try {
newUserRepo.save(user);
} catch (Exception e) {
log.error("新庫寫入失?。簕}", user.getId());
retryQueue.add(user);
}
});
}- 差異定時校驗(yàn)
// 每天凌晨校驗(yàn)差異數(shù)據(jù)
@Scheduled(cron = "0 0 3 * * ?")
public void checkDiff() {
long maxOldId = oldUserRepo.findMaxId();
long maxNewId = newUserRepo.findMaxId();
if (maxOldId != maxNewId) {
log.warn("數(shù)據(jù)主鍵最大不一致,舊庫{} vs 新庫{}", maxOldId, maxNewId);
repairService.fixData();
}
}三、用好工具
不同場景需匹配不同的工具鏈，好比搬家時家具用貨車，細(xì)軟用包裹。
工具選型對照表
工具名稱 | 適用場景 | 10億數(shù)據(jù)速度參考 |
mysqldump | 小型表全量導(dǎo)出 | 不建議（可能天級） |
MySQL Shell | InnoDB并行導(dǎo)出 | 約2-4小時 |
DataX | 多源異構(gòu)遷移 | 依賴資源配置 |
Spark | 跨集群大數(shù)據(jù)量ETL | 30分鐘-2小時 |
Spark遷移核心代碼片段：
val jdbcDF = spark.read
.format("jdbc")
.option("url", "jdbc:mysql://source:3306/db")
.option("dbtable", "users")
.option("partitionColumn", "id")
.option("numPartitions", 100) // 按主鍵切分100個區(qū)
.load()
jdbcDF.write
.format("jdbc")
.option("url", "jdbc:mysql://target:3306/db")
.option("dbtable", "new_users")
.mode(SaveMode.Append)
.save()避坑經(jīng)驗(yàn)：
- 分區(qū)數(shù)量應(yīng)接近Spark執(zhí)行器核數(shù)，太多反而降低效率
- 分區(qū)字段必須是索引列，防止全表掃
四、影子測試
遷移后的數(shù)據(jù)一致性驗(yàn)證，好比宇航員出艙前的模擬訓(xùn)練。
影子庫驗(yàn)證流程：
- 生產(chǎn)流量同時寫入新&舊雙庫（影子庫）
- 對比新舊庫數(shù)據(jù)一致性（抽樣與全量結(jié)合）
- 驗(yàn)證新庫查詢性能指標(biāo)（TP99/TP95延遲）
自動化對比腳本示例：
def check_row_count(old_conn, new_conn):
old_cnt = old_conn.execute("SELECT COUNT(*) FROM users").scalar()
new_cnt = new_conn.execute("SELECT COUNT(*) FROM new_users").scalar()
assert old_cnt == new_cnt, f"行數(shù)不一致: old={old_cnt}, new={new_cnt}"
def check_data_sample(old_conn, new_conn):
sample_ids = old_conn.execute("SELECT id FROM users TABLESAMPLE BERNOULLI(0.1)").fetchall()
for id in sample_ids:
old_row = old_conn.execute(f"SELECT * FROM users WHERE id = {id}").fetchone()
new_row = new_conn.execute(f"SELECT * FROM new_users WHERE id = {id}").fetchone()
assert old_row == new_row, f"數(shù)據(jù)不一致, id={id}"五、回滾
即便做好萬全準(zhǔn)備，也要設(shè)想失敗場景的回滾方案——遷移如跳傘，備份傘必須備好。
回滾預(yù)案關(guān)鍵點(diǎn)：
- 備份快照：遷移前全量快照（物理備份+ Binlog點(diǎn)位）
- 流量回切：準(zhǔn)備路由配置秒級切換舊庫
- 數(shù)據(jù)標(biāo)記：新庫數(shù)據(jù)打標(biāo)，便于清理臟數(shù)據(jù)
快速回滾腳本：
# 恢復(fù)舊庫數(shù)據(jù)
mysql -h舊庫 < backup.sql
# 應(yīng)用Binlog增量
mysqlbinlog --start-positinotallow=154 ./binlog.000001 | mysql -h舊庫
# 切換DNS解析
aws route53 change-resource-record-sets --cli-input-json file://switch_to_old.json總結(jié)
處理10億數(shù)據(jù)的核心心法：
- 分而治之：拆解問題比解決問題更重要。
- 逐步遞進(jìn)：通過灰度驗(yàn)證逐步放大流量。
- 守牢底線：回滾方案必須真實(shí)演練過。
記住——沒有百分百成功的遷移，只有百分百準(zhǔn)備的Plan B！
搬運(yùn)數(shù)據(jù)如同高空走鋼絲，你的安全保障（備份、監(jiān)控、熔斷）就是那根救命繩。
