Thỉnh thoảng phải làm một việc trên rất nhiều dòng:
điền giá trị cho cột mới trên 50 triệu user (backfill);
xoá log cũ hơn 6 tháng;
sửa dữ liệu sai hàng loạt;
import file khách hàng gửi;
export báo cáo.
Viết thì chỉ là một câu SQL, nhưng chạy một câu cho cả bảng trên database đang phục vụ người dùng có thể làm request khác treo, replica chậm theo hàng phút, và không dừng giữa chừng được.
Trang này nói cách chia việc lớn thành nhiều lô nhỏ, cách điều tốc, và cách import, export nhanh. Mọi số liệu được đo thật ngày 2026-10-19 trên MySQL 8.0.46 có một replica và PostgreSQL 17, chạy trong Docker trên laptop, bằng script Node.js (mysql2, pg). Thời gian tuyệt đối phụ thuộc máy; điều đáng nhớ là tỉ lệ giữa các cách làm.
Phép đo: bảng 1 triệu dòng, cập nhật một cột trên mọi dòng. Trong lúc job chạy, một connection khác thử sửa dòng id = 1; một connection đo replica chậm hơn source bao nhiêu. Mỗi ô là khoảng giá trị qua 3 tới 5 lượt chạy:
Cách làm
Thời gian
Lệnh sửa dòng id = 1 phải chờ
Replica chậm tối đa
Số dòng một transaction giữ lock
Một câu UPDATE big SET v = v + 1
3,3 – 4,4 s
3,0 – 4,1 s
5,5 – 9,8 s
~1.000.000
200 lô, mỗi lô 5.000 dòng theo khoá chính
5,1 – 7,8 s
0 – 1 ms
2,6 – 3,8 s
~5.000
200 lô, nghỉ 20 ms giữa các lô
8,4 – 13,6 s
0 – 1 ms
0,2 – 0,5 s, một lần 2,7 s
~5.000
200 lô, nghỉ 50 ms giữa các lô
14,3 – 15,4 s
1 ms
0,23 – 0,24 s
~5.000
200 lô, dừng khi replica chậm quá 500 ms
7,8 – 8,1 s
0 – 1 ms
0,79 – 0,82 s
~5.000
Một câu cho cả bảng gây ra:
Lock giữ tới khi commit. Câu UPDATE lock từng dòng nó sửa và giữ tới cuối. Request nào cần sửa một dòng trong đó phải chờ cả job xong; ở đây là 4 giây, còn với bảng 50 triệu dòng có thể là vài phút, và request web sẽ timeout.
Replica chậm theo cả transaction. Replica chỉ bắt đầu áp dụng khi transaction đã commit trên source, rồi mất chừng ấy thời gian để chạy lại. Trong lúc đó, mọi lần đọc từ replica đều thấy dữ liệu cũ (MySQL chương 9).
Undo log phình to. Mọi phiên bản cũ của 1 triệu dòng phải giữ lại tới khi commit, để còn rollback được và để các transaction khác vẫn đọc được bản cũ (MySQL chương 6).
Lỗi giữa chừng thì mất hết. Lỗi ở dòng thứ 900.000 thì database rollback cả 900.000 dòng, mà rollback cũng tốn thời gian gần bằng lúc chạy. Không có cách tạm dừng rồi chạy tiếp.
Bảng kết quả còn cho thấy hai điều dễ bỏ qua:
Chỉ chia lô thì chưa đủ. Chia lô giải quyết chuyện chờ lock (từ khoảng 4 giây xuống 1 ms), nhưng nếu chạy hết tốc lực thì replica vẫn tụt lại 3 giây. Replica áp dụng chậm hơn source, nên lag cộng dồn qua từng lô.
Nghỉ một khoảng cố định cũng không chắc chắn. Nghỉ 20 ms thường giữ được lag dưới nửa giây, nhưng có một lượt máy bận hơn thì replica vẫn tụt 2,7 giây. Nghỉ 50 ms thì an toàn hơn, đổi lại job chạy gần gấp đôi. Cách tốt nhất là đo lag thật sau mỗi lô và chỉ dừng khi vượt ngưỡng. Với ngưỡng 500 ms, lag luôn quanh ngưỡng đó (cao nhất 0,8 giây do lô đang chạy dở và khoảng lấy mẫu), mà job còn xong nhanh hơn cả hai cách nghỉ cố định, vì nó chỉ dừng khi cần.
Mỗi lô xử lý một khoảng khoá chính, rồi commit. Muốn biết lô tiếp theo bắt đầu từ đâu thì nhớ khoá chính của dòng cuối, giống keyset pagination (Phân trang).
Đừng chia lô bằng OFFSET. Với OFFSET, database phải đọc rồi bỏ qua toàn bộ các dòng đứng trước, nên lô sau luôn chậm hơn lô trước. Phép đo: đọc hết 1 triệu dòng theo lô 5.000:
Cách lấy lô
Lô đầu
Lô thứ 100
Lô cuối
Tổng 200 lô
ORDER BY id LIMIT 5000 OFFSET n
7 ms
43 ms
77 ms
8,7 s
WHERE id > :last ORDER BY id LIMIT 5000
3 ms
3 ms
1 ms
0,42 s
Với bảng 50 triệu dòng, khoảng cách này còn lớn hơn nhiều, vì lô cuối của cách OFFSET phải bỏ qua gần 50 triệu dòng.
Mẫu câu lệnh theo lô:
sql
-- Cập nhật theo khoảng khoá chính (chạy được trên cả MySQL và PostgreSQL)UPDATE users SET email_lower = LOWER(email)WHERE id > :last_id AND id <= :last_id + 5000;-- Xoá theo lô trên MySQL: lặp tới khi 0 dòng bị ảnh hưởngDELETE FROM logs WHERE created_at < '2026-04-01' ORDER BY id LIMIT 5000;-- PostgreSQL không có DELETE … LIMIT (báo lỗi cú pháp); chọn id trước rồi xoáWITH batch AS ( SELECT id FROM logs WHERE created_at < '2026-04-01' ORDER BY id LIMIT 5000)DELETE FROM logs WHERE id IN (SELECT id FROM batch);
Lô bao nhiêu dòng là vừa? Thường là vài nghìn dòng, sao cho mỗi lô chạy dưới khoảng 100 ms. Lô quá nhỏ thì tốn thời gian cho round-trip và commit; lô quá lớn thì lại quay về các vấn đề ở trên. Nếu bảng có khoá chính thưa (nhiều khoảng trống), thì lấy biên trên bằng SELECT MAX(id) FROM (SELECT id FROM users WHERE id > :last ORDER BY id LIMIT 5000) t, để lô nào cũng đủ số dòng.
Xoá rất nhiều dữ liệu cũ theo ngày là việc định kỳ thì nên dùng partition: xoá cả partition (DROP PARTITION) gần như tức thì, thay vì DELETE hàng triệu dòng (MySQL chương 10).
Điều tốc: để replica và người dùng không bị ảnh hưởng
Thử: cập nhật 1 triệu dòng, chia lô và nghỉ giữa các lô
27 msmỗi lô giữ lock (lệnh khác phải chờ tối đa chừng này)
2,2 sreplica chậm tối đa
5,4 sjob chạy xong trên source (200 lô)
Đường biểu diễn replica chậm hơn source bao nhiêu theo thời gian. Không nghỉ thì lag tăng dần tới cuối job, vì replica áp dụng chậm hơn source; nghỉ đủ lâu thì replica theo kịp sau mỗi lô.
Mô hình hiệu chỉnh thô theo số đo trên laptop (MySQL 8.0 + một replica, Docker): source sửa mỗi dòng ~5 µs cộng ~2 ms mỗi lô, replica chậm hơn ~1,4 lần. Số đo thật nằm ở bảng bên dưới trang; mô phỏng chỉ để thấy xu hướng.
Các cách điều tốc, từ đơn giản tới chặt chẽ:
Nghỉ cố định giữa các lô. Đơn giản và thường hiệu quả: trong phép đo ở trên, nghỉ 20 ms làm lag giảm từ khoảng 3 giây xuống dưới nửa giây. Nhưng không chắc chắn: khi máy bận hơn, cùng 20 ms vẫn để replica tụt 2,7 giây. Khoảng nghỉ đúng phụ thuộc vào tải lúc chạy, mà tải thì luôn thay đổi.
Tạm dừng khi replica chậm quá ngưỡng. Sau mỗi lô, đọc lag của replica; vượt ngưỡng (ví dụ 500 ms hay 1 giây) thì chờ tới khi replica theo kịp. Đây là cách nên dùng: trong phép đo, nó giữ lag quanh ngưỡng và còn chạy nhanh hơn nghỉ cố định, vì lúc replica theo kịp thì không phải nghỉ. Seconds_Behind_Source trong SHOW REPLICA STATUS chỉ có độ chính xác tới giây và có lúc ra NULL. Chính xác hơn là dùng một bảng heartbeat, như cách phép đo trên trang này làm: source ghi thời gian hiện tại mỗi 100 ms, đọc bảng đó trên replica rồi lấy hiệu.
Tạm dừng khi database bận, ví dụ số thread đang chạy (Threads_running) vượt ngưỡng, hoặc tránh giờ cao điểm.
Công cụ đổi schema online gh-ost làm đúng các việc trên khi sao chép bảng. Nó tự sinh heartbeat để đo lag. Tuỳ chọn --max-lag-millis dừng sao chép khi replica chậm quá ngưỡng, --max-load dừng khi các chỉ số tải vượt ngưỡng, và một file cờ cho phép dừng bằng tay. Trong lúc dừng, gh-ost ngừng ghi dữ liệu để giảm tải cho primary (gh-ost: Throttle).
Job chạy hàng giờ thì sớm muộn cũng bị dừng: deploy mới, mất kết nối, bị kill vì quá tải. Job phải chạy lại được mà không làm sai dữ liệu:
Mỗi lô idempotent: chạy lại một lô đã xong cũng không gây hại. Ví dụ thêm điều kiện AND email_lower IS NULL, hoặc ghi giá trị tuyệt đối thay vì v = v + 1.
Lưu tiến độ (khoá chính cuối cùng đã xong) vào một bảng, ghi cùng transaction với lô đó, để tiến độ và dữ liệu không bao giờ lệch nhau.
Chỉ một instance chạy job cùng lúc, ví dụ dùng advisory lock (Distributed lock).
Code Node.js: job backfill có điều tốc và chạy lại được
js
// mysql2/promise. Điền cột email_lower cho bảng users, mỗi lô 5.000 dòng.const JOB = 'backfill_users_email_lower'const sleep = (ms) => new Promise((r) => setTimeout(r, ms))async function replicaLagSeconds(replica) { const [[st]] = await replica.query('SHOW REPLICA STATUS') return st?.Seconds_Behind_Source ?? Infinity // NULL: replica đang dừng, coi như chậm vô hạn}export async function backfill(pool, replica, { batch = 5000, pauseMs = 20, maxLagSeconds = 1 } = {}) { const [[row]] = await pool.query('SELECT last_id FROM job_progress WHERE job = ?', [JOB]) let lastId = row?.last_id ?? 0 const [[{ maxId }]] = await pool.query('SELECT MAX(id) AS maxId FROM users') while (lastId < maxId) { const to = lastId + batch const conn = await pool.getConnection() try { await conn.beginTransaction() await conn.query( 'UPDATE users SET email_lower = LOWER(email) WHERE id > ? AND id <= ? AND email_lower IS NULL', [lastId, to]) await conn.query( 'INSERT INTO job_progress (job, last_id) VALUES (?, ?) AS new ON DUPLICATE KEY UPDATE last_id = new.last_id', [JOB, to]) await conn.commit() } catch (err) { await conn.rollback() throw err // chạy lại job sẽ tiếp tục từ lastId đã lưu } finally { conn.release() } lastId = to await sleep(pauseMs) while ((await replicaLagSeconds(replica)) > maxLagSeconds) await sleep(1000) }}
Bảng job_progress(job VARCHAR(100) PRIMARY KEY, last_id BIGINT NOT NULL) giữ tiến độ. Muốn chạy song song cho nhanh thì chia khoảng khoá chính cho vài worker; nhưng mỗi worker thêm là thêm tải cho source và replica, nên vẫn phải điều tốc chung.
Đổi cột trên bảng đang chạy (đổi tên, đổi kiểu, tách cột) không thể làm trong một bước mà không có downtime. Cách an toàn là expand – contract, và bước backfill ở giữa chính là job theo lô ở trên:
Phép đo: import 20.000 dòng (bảng có khoá chính và một index phụ):
Cách
MySQL
PostgreSQL
Từng dòng, mỗi dòng tự commit (autocommit)
51,4 s
7,6 s
Từng dòng, cả 20.000 dòng trong một transaction
4,1 s
6,2 s
INSERT 1.000 dòng mỗi câu
0,26 s
0,20 s
Và 200.000 dòng, so INSERT nhiều dòng với công cụ nạp file của từng database (file CSV đã nằm sẵn trên máy chủ database):
Cách
MySQL
PostgreSQL
INSERT 1.000 dòng mỗi câu
2,25 s
1,55 s
LOAD DATA INFILE (MySQL) / COPY … FROM (PostgreSQL)
0,79 s
0,48 s
Rút ra:
Đừng commit từng dòng. Ở chế độ autocommit, InnoDB ghi log xuống đĩa sau mỗi lệnh INSERT (MySQL: Bulk Data Loading for InnoDB). Gom vào transaction đã nhanh hơn khoảng 12 lần trên MySQL. MySQL chậm hơn PostgreSQL nhiều ở cách này; một lý do có thể là mỗi commit trên MySQL phải ghi cả redo log lẫn binlog, mà phép đo này có bật binlog cho replica (MySQL chương 8).
Gửi nhiều dòng trong một câu. Mỗi câu lệnh tốn chi phí gửi, phân tích cú pháp, khứ hồi mạng (MySQL: Optimizing INSERT Statements). Nhanh hơn từng dòng tự commit khoảng 200 lần trên MySQL và gần 40 lần trên PostgreSQL.
Có file thì dùng công cụ nạp file: nhanh hơn INSERT nhiều dòng khoảng 3 lần trong phép đo này. Tài liệu MySQL viết LOAD DATA “thường nhanh hơn khoảng 20 lần” so với INSERT; con số này so với INSERT thông thường, không phải INSERT 1.000 dòng mỗi câu như ở đây. Tài liệu PostgreSQL khuyên dùng COPY vì nó được tối ưu cho việc nạp nhiều dòng (PostgreSQL: Populating a Database). Với file từ máy khác, MySQL có LOAD DATA LOCAL INFILE (phải bật local_infile), còn PostgreSQL có \copy của psql hoặc COPY … FROM STDIN qua driver.
Thêm các mẹo từ tài liệu chính thức, dùng khi nạp bảng mới hoặc bảng không phục vụ người dùng trong lúc nạp:
MySQL InnoDB
PostgreSQL
Transaction
Tắt autocommit, commit một lần hoặc theo từng lô lớn
Phép đo: export 1 triệu dòng ra file CSV 86 MB bằng Node.js, mỗi cách chạy hai lần:
Cách
Bộ nhớ process (RSS) lớn nhất
Heap lớn nhất
Thời gian
const [rows] = await conn.query(sql) rồi ghi file
351 – 359 MB
214 – 217 MB
1,0 – 1,4 s
Stream từng dòng ra file
127 MB
32 – 35 MB
0,8 s
Cách tải hết giữ cả 1 triệu object trong bộ nhớ cùng lúc; bảng 10 triệu dòng sẽ làm process Node.js hết bộ nhớ. Stream giữ bộ nhớ gần như không đổi dù bảng lớn tới đâu, và còn nhanh hơn vì vừa đọc vừa ghi:
js
import mysql from 'mysql2' // stream dùng API callback, không phải mysql2/promiseimport fs from 'node:fs'import { pipeline } from 'node:stream/promises'import { Transform } from 'node:stream'const conn = mysql.createConnection(config)const toCsv = new Transform({ objectMode: true, transform(row, _enc, cb) { cb(null, `${row.id},${row.email}\n`) }})await pipeline(conn.query('SELECT id, email FROM users').stream(), toCsv, fs.createWriteStream('users.csv'))conn.end()
pipeline lo luôn back-pressure: file ghi chậm thì stream tạm ngừng đọc từ database. Cũng có một lưu ý: cả lần export là một câu SELECT chạy suốt. Trên InnoDB, nó giữ một snapshot đọc trong suốt thời gian đó, nên undo log không dọn được. Export kéo dài hàng giờ thì nên đọc theo lô keyset (WHERE id > :last ORDER BY id LIMIT 10000), mỗi lô là một câu ngắn. Đổi lại, các lô không cùng một snapshot, tức là dữ liệu có thể đổi giữa các lô.
Chạy thử trên bản sao dữ liệu thật (staging, hoặc bản restore từ backup), đo thời gian một lô.
Mỗi lô có khoá chính trong điều kiện, có index phù hợp, chạy dưới khoảng 100 ms.
Có điều tốc: nghỉ giữa các lô, dừng khi replica chậm quá ngưỡng.
Mỗi lô idempotent; tiến độ lưu cùng transaction; chỉ một instance chạy.
Có cách dừng ngay (cờ trong bảng cấu hình, tín hiệu, file cờ) và theo dõi được tiến độ.
Biết đường lui: backup trước đó, hoặc giữ dữ liệu cũ (cột cũ, bảng cũ) tới khi kiểm tra xong.
Ghi nhớ
Một câu cho cả bảng giữ lock tới khi commit (đo: lệnh khác chờ ~4 s), làm replica chậm theo cả transaction, phình undo log, và không dừng được giữa chừng.
Chia lô theo khoá chính bằng keyset, không dùng OFFSET (đo: 0,42 s so với 8,7 s cho 1 triệu dòng). PostgreSQL không có DELETE … LIMIT, dùng CTE chọn id trước.
Chia lô chưa đủ: chạy hết tốc lực thì replica vẫn tụt khoảng 3 s. Nghỉ cố định cũng không chắc chắn (20 ms có lần vẫn tụt 2,7 s). Đo lag sau mỗi lô và dừng khi vượt ngưỡng, như gh-ost làm: lag giữ quanh ngưỡng mà job còn nhanh hơn.
Job phải chạy lại được: lô idempotent, lưu tiến độ cùng transaction, chỉ một instance chạy.
Import: không commit từng dòng; INSERT nhiều dòng một câu; có file thì LOAD DATA / COPY (đo: từ 51 s xuống 0,26 s cho 20.000 dòng, rồi nhanh thêm khoảng 3 lần).
Export trong Node.js: stream thay vì tải hết (đo: 127 MB so với 355 MB RSS cho 1 triệu dòng); export rất dài thì đọc theo lô keyset.
Đã đo ngày 2026-10-19 trên MySQL 8.0.46 (source và một replica GTID, cấu hình mặc định, binlog bật) và PostgreSQL 17, chạy Docker trên laptop, bằng script Node.js dùng mysql2 3.24 và pg 8.23. Các phép đo:
Một câu so với chia lô: bảng 1 triệu dòng. Heartbeat ghi thời gian vào bảng hb trên source mỗi 100 ms, đọc trên replica mỗi 200 ms để tính lag. Số dòng bị lock lấy từ information_schema.INNODB_TRX.trx_rows_locked. Lệnh thử sửa dòng id = 1 gửi 300 ms sau khi job bắt đầu.
OFFSET so với keyset: đọc 1 triệu dòng theo lô 5.000.
Import: 20.000 và 200.000 dòng; file CSV cho LOAD DATA / COPY tạo sẵn trên máy chủ database bằng SELECT … INTO OUTFILE / COPY … TO.
Export: RSS và heap lấy mẫu mỗi 20 ms bằng process.memoryUsage().
DELETE … LIMIT trên PostgreSQL: chạy bằng psql.
Lần đo đầu có hai chỗ sai do script: lag tính sai khi replica còn giá trị khởi tạo ts = 0, và đo nhầm trx_rseg_history_len (chỉ đếm undo của transaction đã commit). Cả hai đã được sửa trước khi lấy số. Mô phỏng điều tốc là mô hình hiệu chỉnh thô theo số đo, không phải số đo.
Script nằm trong thư mục lab/xu-ly-theo-lo/ của repo. Tự chạy lại bằng npm run lab:up rồi npm run lab -- xu-ly-theo-lo (cách chạy).
batch-update.mjs: một câu so với chia lô, đo chờ lock và replica lag
js
// Trang database/xu-ly-theo-lo: một câu UPDATE cả bảng so với chia lô theo khoá chính, trên MySQL có replica.// Trong lúc job chạy:// - heartbeat: ghi thời gian vào bảng hb trên source mỗi 100 ms; đọc hb trên replica mỗi 200 ms để tính lag;// - probe: 300 ms sau khi job bắt đầu, thử UPDATE dòng id = 1 và đo phải chờ lock bao lâu;// - monitor: số dòng một transaction đang giữ lock (INNODB_TRX.trx_rows_locked).import mysql from 'mysql2/promise'import { MYSQL, REPLICA, sleep } from '../common/config.mjs'const SRC = { ...MYSQL, database: 'batch' }const REP = { ...REPLICA, database: 'batch' }const N = 1_000_000async function setup() { const c = await mysql.createConnection({ ...MYSQL, multipleStatements: true }) await c.query(`CREATE DATABASE IF NOT EXISTS batch; USE batch; DROP TABLE IF EXISTS big; DROP TABLE IF EXISTS hb; CREATE TABLE hb (id INT PRIMARY KEY, ts BIGINT NOT NULL); INSERT INTO hb VALUES (1, 0); CREATE TABLE big (id INT PRIMARY KEY, v INT NOT NULL, note VARCHAR(100) NOT NULL); SET SESSION cte_max_recursion_depth = 2000000; INSERT INTO big WITH RECURSIVE s AS (SELECT 1 AS n UNION ALL SELECT n + 1 FROM s WHERE n < ${N}) SELECT n, 0, REPEAT('x', 80) FROM s;`) await c.end() // chờ replica có đủ dữ liệu const r = await mysql.createConnection(REPLICA) for (;;) { const [[x]] = await r.query('SELECT COUNT(*) AS n FROM batch.big').catch(() => [[{ n: 0 }]]) if (x.n === N) break await sleep(500) } await r.end()}async function measure(label, work) { const hbConn = await mysql.createConnection(SRC) const repConn = await mysql.createConnection(REP) const probe = await mysql.createConnection(SRC) const mon = await mysql.createConnection(SRC) await probe.query('SET SESSION innodb_lock_wait_timeout = 120') let stop = false, maxLag = 0, maxLocked = 0 const lags = [] const hb = (async () => { while (!stop) { await hbConn.query('UPDATE hb SET ts = ? WHERE id = 1', [Date.now()]); await sleep(100) } })() const rep = (async () => { while (!stop) { const [[x]] = await repConn.query('SELECT ts FROM hb WHERE id = 1') if (x.ts > 0) { // ts = 0 là giá trị khởi tạo, heartbeat chưa sang tới replica const lag = Date.now() - x.ts lags.push(lag); if (lag > maxLag) maxLag = lag } await sleep(200) } })() const locked = (async () => { while (!stop) { const [[x]] = await mon.query('SELECT COALESCE(MAX(trx_rows_locked), 0) AS n FROM information_schema.INNODB_TRX') if (Number(x.n) > maxLocked) maxLocked = Number(x.n) await sleep(200) } })() await sleep(1500) const t0 = Date.now() const workP = work() await sleep(300) const p0 = Date.now() await probe.query('UPDATE big SET note = note WHERE id = 1') const probeWait = Date.now() - p0 await workP const dur = Date.now() - t0 // chờ replica đuổi kịp hẳn rồi mới dừng đo for (let i = 0; i < 600; i++) { await sleep(200); if (lags.length && lags[lags.length - 1] < 400) break } stop = true await Promise.all([hb, rep, locked]) for (const c of [hbConn, repConn, probe, mon]) await c.end() console.log(`${label.padEnd(44)} thời gian ${String(dur).padStart(5)} ms · chờ lock ${String(probeWait).padStart(5)} ms · replica chậm tối đa ${String(maxLag).padStart(5)} ms · lock tối đa ${maxLocked} dòng`)}await setup()await measure('Một câu UPDATE cả bảng', async () => { const c = await mysql.createConnection(SRC) await c.query('UPDATE big SET v = v + 1') await c.end()})await measure('Chia lô 5.000 dòng theo khoá chính', async () => { const c = await mysql.createConnection(SRC) for (let from = 1; from <= N; from += 5000) await c.query('UPDATE big SET v = v + 1 WHERE id BETWEEN ? AND ?', [from, from + 4999]) await c.end()})for (const pause of [20, 50]) { await measure(`Chia lô 5.000 dòng, nghỉ ${pause} ms giữa các lô`, async () => { const c = await mysql.createConnection(SRC) for (let from = 1; from <= N; from += 5000) { await c.query('UPDATE big SET v = v + 1 WHERE id BETWEEN ? AND ?', [from, from + 4999]) await sleep(pause) } await c.end() })}// Điều tốc theo lag thật: sau mỗi lô đọc heartbeat trên replica, chậm quá 500 ms thì chờ tới khi theo kịp.await measure('Chia lô 5.000 dòng, dừng khi replica chậm > 500 ms', async () => { const c = await mysql.createConnection(SRC) const r = await mysql.createConnection(REP) const lag = async () => { const [[x]] = await r.query('SELECT ts FROM hb WHERE id = 1'); return Date.now() - x.ts } for (let from = 1; from <= N; from += 5000) { await c.query('UPDATE big SET v = v + 1 WHERE id BETWEEN ? AND ?', [from, from + 4999]) while ((await lag()) > 500) await sleep(50) } await c.end() await r.end()})process.exit(0)
offset-keyset.mjs: chia lô bằng OFFSET so với keyset
js
// Đọc 1 triệu dòng (bảng batch.big do batch-update.mjs tạo) theo lô 5.000 bằng OFFSET và bằng keyset.import mysql from 'mysql2/promise'import { MYSQL } from '../common/config.mjs'const ms = (t) => Math.round(Number(process.hrtime.bigint() - t) / 1e6)const now = () => process.hrtime.bigint()const c = await mysql.createConnection({ ...MYSQL, database: 'batch' })for (const mode of ['offset', 'keyset']) { const times = [] let last = 0, rows = 0 const t0 = now() for (let i = 0; ; i++) { const t = now() const [batch] = mode === 'offset' ? await c.query('SELECT id, v FROM big ORDER BY id LIMIT 5000 OFFSET ?', [i * 5000]) : await c.query('SELECT id, v FROM big WHERE id > ? ORDER BY id LIMIT 5000', [last]) times.push(ms(t)) if (!batch.length) break rows += batch.length last = batch[batch.length - 1].id } console.log(`${mode.padEnd(7)} ${rows} dòng, ${times.length - 1} lô · lô đầu ${times[0]} ms · lô thứ 100 ${times[100]} ms · lô cuối ${times[times.length - 2]} ms · tổng ${ms(t0)} ms`)}await c.end()
import.mjs: import từng dòng, nhiều dòng một câu, LOAD DATA, COPY
js
// Tốc độ import vào MySQL và PostgreSQL (bảng có khoá chính và một index phụ):// 20.000 dòng: từng dòng tự commit, từng dòng trong một transaction, 1.000 dòng mỗi câu INSERT;// 200.000 dòng: INSERT 1.000 dòng mỗi câu so với LOAD DATA INFILE / COPY FROM// (file CSV tạo sẵn trên máy chủ database bằng SELECT … INTO OUTFILE / COPY … TO, nên chỉ đo phần nạp).import mysql from 'mysql2/promise'import pg from 'pg'import { spawnSync } from 'node:child_process'import { MYSQL, PG } from '../common/config.mjs'import { ensureDb } from '../common/run-sql.mjs'const ms = (t) => Math.round(Number(process.hrtime.bigint() - t) / 1e6)const now = () => process.hrtime.bigint()const row = (i) => [i, `user${i}@x.vn`, i % 1000]ensureDb('pg', 'batch')function mysqlOps(c) { return { reset: async () => { await c.query('DROP TABLE IF EXISTS imp'); await c.query('CREATE TABLE imp (id INT PRIMARY KEY, email VARCHAR(50) NOT NULL, score INT NOT NULL, KEY (email))') }, one: (i) => c.query('INSERT INTO imp VALUES (?, ?, ?)', row(i)), begin: () => c.beginTransaction(), commit: () => c.commit(), many: (from, to) => { const b = []; for (let i = from; i < to; i++) b.push(row(i)); return c.query('INSERT INTO imp VALUES ?', [b]) }, dump: () => c.query("SELECT id, email, score FROM imp INTO OUTFILE '/var/lib/mysql-files/imp.csv' FIELDS TERMINATED BY ','"), load: () => c.query("LOAD DATA INFILE '/var/lib/mysql-files/imp.csv' INTO TABLE imp FIELDS TERMINATED BY ','"), count: async () => (await c.query('SELECT COUNT(*) AS n FROM imp'))[0][0].n }}function pgOps(c) { return { reset: async () => { await c.query('DROP TABLE IF EXISTS imp'); await c.query('CREATE TABLE imp (id INT PRIMARY KEY, email TEXT NOT NULL, score INT NOT NULL)'); await c.query('CREATE INDEX ON imp (email)') }, one: (i) => c.query('INSERT INTO imp VALUES ($1, $2, $3)', row(i)), begin: () => c.query('BEGIN'), commit: () => c.query('COMMIT'), many: (from, to) => { const vals = [], params = [] for (let i = from; i < to; i++) { const b = params.length; params.push(...row(i)); vals.push(`($${b + 1}, $${b + 2}, $${b + 3})`) } return c.query(`INSERT INTO imp VALUES ${vals.join(',')}`, params) }, dump: () => c.query("COPY imp TO '/tmp/imp.csv' WITH (FORMAT csv)"), load: () => c.query("COPY imp FROM '/tmp/imp.csv' WITH (FORMAT csv)"), count: async () => (await c.query('SELECT COUNT(*)::int AS n FROM imp')).rows[0].n }}async function bench(name, ops) { const R = 20000 await ops.reset(); let t = now() for (let i = 1; i <= R; i++) await ops.one(i) const auto = ms(t) await ops.reset(); t = now() await ops.begin(); for (let i = 1; i <= R; i++) await ops.one(i); await ops.commit() const trx = ms(t) await ops.reset(); t = now() for (let s = 1; s <= R; s += 1000) await ops.many(s, s + 1000) const multi = ms(t) console.log(`${name.padEnd(10)} 20.000 dòng · từng dòng tự commit ${auto} ms · từng dòng trong một transaction ${trx} ms · 1.000 dòng mỗi câu ${multi} ms`) const B = 200000 await ops.reset(); t = now() for (let s = 1; s <= B; s += 1000) await ops.many(s, s + 1000) const multiBig = ms(t) await ops.dump() await ops.reset(); t = now() await ops.load() const load = ms(t) console.log(`${name.padEnd(10)} 200.000 dòng · 1.000 dòng mỗi câu ${multiBig} ms · nạp file ${load} ms (nạp được ${await ops.count()} dòng)`)}const m = await mysql.createConnection({ ...MYSQL, database: 'batch' })// SELECT … INTO OUTFILE không ghi đè file đã có, nên xoá file của lần chạy trướcspawnSync('docker', ['exec', 'lab-mysql', 'rm', '-f', '/var/lib/mysql-files/imp.csv'])await bench('MySQL', mysqlOps(m))await m.end()const p = new pg.Client({ ...PG, database: 'batch' }); await p.connect()await bench('PostgreSQL', pgOps(p))await p.end()
export.mjs: export bằng cách tải hết so với stream
js
// Export 1 triệu dòng (bảng batch.big) ra file CSV trong Node.js.// node xu-ly-theo-lo/export.mjs all tải hết kết quả vào bộ nhớ rồi ghi// node xu-ly-theo-lo/export.mjs stream đọc từng dòng từ mysql2 và ghi dần ra file// Đo RSS và heap lớn nhất (lấy mẫu mỗi 20 ms). Mỗi chế độ nên chạy trong một process riêng.import mysql from 'mysql2'import fs from 'node:fs'import os from 'node:os'import path from 'node:path'import { pipeline } from 'node:stream/promises'import { Transform } from 'node:stream'import { MYSQL } from '../common/config.mjs'const mode = process.argv[2]let maxRss = 0, maxHeap = 0const sampler = setInterval(() => { const m = process.memoryUsage() if (m.rss > maxRss) maxRss = m.rss if (m.heapUsed > maxHeap) maxHeap = m.heapUsed}, 20)const conn = mysql.createConnection({ ...MYSQL, database: 'batch' })const out = path.join(os.tmpdir(), `lab-export-${mode}.csv`)const t0 = Date.now()if (mode === 'all') { const [rows] = await conn.promise().query('SELECT id, v, note FROM big') const f = fs.createWriteStream(out) for (const r of rows) if (!f.write(`${r.id},${r.v},${r.note}\n`)) await new Promise((ok) => f.once('drain', ok)) await new Promise((ok) => f.end(ok))} else { const toCsv = new Transform({ objectMode: true, transform(r, _e, cb) { cb(null, `${r.id},${r.v},${r.note}\n`) } }) await pipeline(conn.query('SELECT id, v, note FROM big').stream(), toCsv, fs.createWriteStream(out))}clearInterval(sampler)conn.end()const size = fs.statSync(out).sizefs.unlinkSync(out)const mb = (x) => Math.round(x / 1048576)console.log(`${mode.padEnd(7)} ${Date.now() - t0} ms · RSS lớn nhất ${mb(maxRss)} MB · heap lớn nhất ${mb(maxHeap)} MB · file ${mb(size)} MB`)
delete-pg.sql: xoá theo lô trên PostgreSQL
sql
-- PostgreSQL không có DELETE … LIMIT; xoá theo lô bằng CTE chọn id trước.DROP TABLE IF EXISTS logs;CREATE TABLE logs (id INT PRIMARY KEY, created_at DATE NOT NULL);INSERT INTO logs SELECT g, DATE '2026-01-01' + (g % 300) FROM generate_series(1, 100000) g;SELECT COUNT(*) AS old_rows_before FROM logs WHERE created_at < '2026-07-01';\echo '--- DELETE … ORDER BY … LIMIT: lỗi cú pháp'DELETE FROM logs WHERE created_at < '2026-07-01' ORDER BY id LIMIT 5000;\echo '--- CTE chọn 5.000 id rồi xoá'WITH batch AS (SELECT id FROM logs WHERE created_at < '2026-07-01' ORDER BY id LIMIT 5000)DELETE FROM logs WHERE id IN (SELECT id FROM batch);SELECT COUNT(*) AS old_rows_after FROM logs WHERE created_at < '2026-07-01';