Large data migrations: moving a live system's data with no downtime

The interviewer says: "Your payments database is on one old server and must move to four new shards. It holds 900 GB, it takes writes all day, and you may not stop the service. How do you move it?"

A weak answer says "copy it at night" or "write to both places" and stops. A strong answer gives an order of steps. It also says how you check that old and new agree, how you go back, and how long each step takes in numbers.

This is an SDE-3 question.

Think of moving a shop to a new building while customers keep buying. You carry the stock across in trips. During the trips, new sales happen, so you also note every sale and update the new shelves. At last you check that both shops hold the same stock, send a few customers to the new shop, and then all of them. You keep the old keys for a week, in case you must go back.

A migration moves data from an old place to a new place while users keep writing. The place can be a new database, a new shard layout (lesson 69) or a new region. Lesson 41 changed the shape of one table. Lesson 120 moved code piece by piece. This lesson moves the data of a whole system.

The standard order

1. Backfill. Copy a snapshot, which is the data at one moment, from the old place to the new. Remember the log position at that moment.
2. Follow the changes. Every change made after the snapshot must also reach the new place. There are two ways. With change data capture (lesson 92), a program reads the old database's log and applies each change to the new one. With dual writes, the application writes to both places. Dual writes have the problem of lesson 92: two places cannot commit together, so they can disagree. Change data capture keeps the old place as the only writer.
3. Check. Compare old and new. Count rows. Compare a checksum, a short fingerprint of the rows, for each chunk of rows. Run shadow reads: for some real requests, read from both places, compare the answers, and return the old answer.
4. Move reads. Send reads to the new place a few percent at a time. Pick users by a hash of the user ID, as in lesson 127.
5. Move writes. The new place becomes the one that takes writes. This is the risky step.
6. Keep a way back. After the switch, send changes in the other direction, from new to old, so the old place stays current. Then you can return.
7. Delete the old copy last. Only this step cannot be undone.

Vitess, a system for scaling MySQL, documents the same flow for moving tables: create (copy), check with VDiff, switch traffic, reverse the traffic if needed, and complete.

This is the expand and contract of lesson 41, for a whole system: add the new place (expand), move users to it, then remove the old place (contract).

A job that can stop and start again

The copy takes hours, and it will be stopped: by a crash, a deploy or an alarm. So it must be safe to run again (lesson 89). Write each row with an "upsert" (insert or update), so the same row twice gives the same result. Save a checkpoint, the last key copied, so a restart continues from there. Also throttle: slow down when production is busy.

def copy_chunk(last_id):
    rows = old.read("SELECT * FROM ledger WHERE id > %s ORDER BY id LIMIT 5000", last_id)
    if not rows:
        return None                  # the copy is finished
    for r in rows:
        new.upsert_if_newer(r)       # the same row twice gives the same result
    save_checkpoint(rows[-1].id)     # after a stop, start again from here
    while old.p99_ms() > 120:        # throttle: wait while production is slow
        sleep(1)
    return rows[-1].id

What goes wrong

• A delete during the copy. The copy reads a row. The user deletes it. The change arrives at the new place before the copy writes the row, and the row comes back. The fix: every write carries a version, and a write wins only if it is newer. A delete leaves a small marker, a tombstone, until the copy has ended and the check is clean.
• Order. Two changes to one row must be applied in the order they happened. Keep changes to one key in order, as a log does for a partition key (lesson 87).
• A hot row. A row that changes many times each second never matches in a checksum taken while it moves. Compare recent rows after they settle.
• A copy that never catches up. The new place must apply changes faster than they arrive. Otherwise the lag, the number of changes waiting, only grows.
• A log that is too short. Some databases, such as MySQL with its binary log, delete the log after a set time. If the copy takes longer, the changes after the snapshot are gone, and you must start again. Keep the log longer, or start a reader at the snapshot that copies the changes into a queue as they happen. A Postgres replication slot (lesson 92) keeps the log until it is read, so there the risk is a full disk.

The switch for writes

This step needs a short pause. Do it in this order.

1. Hold new writes for a few seconds, in the application or a proxy. Reads continue.
2. Wait until the lag is zero, then run a last check.
3. Change the routing so that writes go to the new place.
4. Start the flow from new to old, then let the held writes through.

Going back is the same steps in the other direction. It works only while the old place can hold everything that the new place now accepts. If the new place gets a column or a rule that the old place cannot hold, going back ends there, so decide when that point comes.

The same idea for one table

Tools such as gh-ost (GitHub) change the shape of one big MySQL table in the same order. They create a ghost table with the new shape, copy rows into it in small chunks, apply new changes read from the binary log, and slow down when the database is busy. At the end they swap the two table names in a very short step. Older tools use triggers instead of the log, which add work to every write (see Sources).

Planning with numbers

• Copy time = data size / copy speed.
• Changes waiting at the end of the copy = changes per second x copy time.
• Catch-up time = changes waiting / (apply speed - arrival speed).
• Lag in seconds = changes waiting / apply speed. Switch writes only when the lag is a second or two.

Compared with a window and a rewrite

A maintenance window stops writes, copies, and starts again: simple, but the downtime is the whole copy time. A rewrite builds new code and often a new data model, and switches once. A strangler (lesson 120) moves a slice of code, and the data of that slice must move too, with the steps above. Online migration costs weeks of calendar time and engineer work, and no downtime.

ChaiPay moves a ledger table to four shards. Assumptions (made up to teach with): 1,800,000,000 rows of 500 bytes. The old database can give 40 MB per second to the copy without hurting users, and 200 MB per second at most. Writes average 1,200 per second and peak at 3,000. Reads average 6,000 per second. A change in the log is about 500 bytes. The new place applies 5,000 changes per second. The old database deletes its log after 6 hours. The service goal is 99.95%.

Step 1: copy time (the predict box). Size = 1,800,000,000 x 500 bytes = 900 GB = 900,000 MB. At 40 MB per second: 900,000 / 40 = 22,500 seconds = 6.25 hours. At 200 MB per second it would take 4,500 seconds = 75 minutes.

Step 2: changes waiting. 1,200 x 22,500 = 27,000,000 changes. At 500 bytes each, that is 13.5 GB of log.

Step 3: the log. The copy takes 22,500 seconds, but the log is kept for 6 x 3,600 = 21,600 seconds. The log is too short by 900 seconds. Raise the log time to 24 hours before you start, or copy the log into a queue from the start. Twenty-four hours of log is 1,200 x 86,400 x 500 bytes = 51.8 GB of disk.

Step 4: catch-up. The new place applies 5,000 and 1,200 arrive, so the backlog falls by 3,800 per second: 27,000,000 / 3,800 = 7,105 seconds = 2 hours. At the peak of 3,000 it still falls. If the new place applied only 1,000 per second, the backlog would grow by 1,200 - 1,000 = 200 per second, 720,000 each hour, and never reach zero. With 10,000 changes waiting, the lag is 10,000 / 5,000 = 2 seconds.

Step 5: check. A chunk is 100,000 rows, so 1,800,000,000 / 100,000 = 18,000 chunks to compare. Shadow-read 10% of 6,000 reads per second: 600 comparisons per second. Assumption: 0.02% differ, mostly rows that just changed: 0.12 per second, 432 an hour. Rule: no unexplained difference for 24 hours.

Step 6: reads and writes. Move reads to 1%, 5%, 25% and 50%, one day each, then to 100%. Switch writes with a pause of seconds, start the flow from new to old, and keep the old copy for 7 days. The plan takes about 6.25 hours + 2 hours + 1 day + 4 days + 7 days = 12 days.

Step 7: the window. Copying with writes stopped, at full speed, takes 75 minutes. A 99.95% goal allows 43,200 x 0.0005 = 21.6 minutes a month. The window does not fit.

What companies wrote

• Stripe wrote on 2 Feb 2017 about moving subscription data online in four steps: write to both places, change reads, change writes, remove the old data. It compared old and new reads in production with a library before it switched. Link: https://stripe.com/blog/online-migrations (opens in a new tab)
• Figma wrote on 14 Mar 2024 that its first horizontally sharded table went live in September 2023, after about nine months of work. It said it wanted to avoid double-writes. The failover had ten seconds of partial availability on primaries (the company's figures). Link: https://www.figma.com/blog/how-figmas-databases-team-lived-to-tell-the-scale/ (opens in a new tab)
• Zerodha listed on 29 Jul 2024 a move of about 3 TB of its client database from Postgres 11 to Postgres 16, with six months shown for research, testing and go-live. The page does not describe the method. Link: https://zerodha.com/z-connect/business-updates/zerodhas-infrastructure-upgrades (opens in a new tab)

An online migration solves downtime: 0 minutes against 75, which would break a 99.95% goal. It costs weeks (12 days here), a log kept for 24 hours, a compare step, and a way back that must be built. I would take a window when the copy fits inside the allowed downtime, for example a small database. I would take dual writes when I cannot read the old log, and change data capture when I can.

"Copy at night, then switch in the morning." At 1,200 changes per second, the copy leaves 27,000,000 changes waiting. Without following them, the new place is wrong.

"If the counts match, the data matches." A changed row keeps the count. Use checksums and shadow reads.

"Delete the old database once the switch works." The way back needs it. In step 6 it stays for 7 days.

1. List the standard order of steps. What is a shadow read?
2. A database of 600 GB is copied at 30 MB per second. Writes average 800 per second, and the new place applies 2,000 per second. (a) How long is the copy? (b) How many changes are waiting at the end? (c) How long to catch up?
3. The log is kept for 12 hours and the copy needs 20 hours. What do you do?