MySQL数据归档之从中小表到大表的全场景方案

-- 典型组合
where status in ('closed', 'cancelled')
  and create_time < @cutoff   -- 如 90 天前
  and keep_flag = 0

数据量 / 特征                    推荐路径
─────────────────────────────────────────────────────────
s  < 500 万、非分区、日增可控      → 分批 delete + 归档库
m  500 万～5000 万、可改表结构     → 时间分区 + exchange/drop
l  单表亿级、短期难分区           → 热数据重建 + 双表 rename
xl 按月暴涨、整月可封闭           → 整表 rename 轮换

阶段 1：s 档 — 脚本分批 delete + 归档库（快速上线）
阶段 2：m 档 — 在线表加分区，新方案切分区归档
阶段 3：l 档 — 历史膨胀后做一次 rename 瘦身，之后走分区

定 cutoff → 分批选取 → insert 归档库 → 校验 → delete 源表 → 记日志 → 全局对账

start transaction;

insert into archive_db.orders_archive
  (id, user_id, amount, status, create_time, archive_batch_id)
select id, user_id, amount, status, create_time, '20240818_001'
from prod_db.orders
where status = 'closed'
  and create_time < '2024-05-01 00:00:00'
  and keep_flag = 0
  and id > 1000000 and id <= 1010000;

delete from prod_db.orders
where status = 'closed'
  and create_time < '2024-05-01 00:00:00'
  and keep_flag = 0
  and id > 1000000 and id <= 1010000;

commit;

create table orders (
  id bigint not null,
  create_time datetime not null,
  ...
  primary key (id, create_time)  -- 分区键必须进主键/唯一键
) partition by range (to_days(create_time)) (
  partition p202401 values less than (to_days('2024-02-01')),
  partition p202402 values less than (to_days('2024-03-01')),
  ...
  partition p_future values less than maxvalue
);

create table orders_archive_202401 like orders;

alter table orders
  exchange partition p202401
  with table orders_archive_202401;

alter table orders drop partition p202401;

1. 把「要保留的热数据」复制到 orders_temp
2. rename orders        → orders_backup_20240818   （整表变备份，含全量快照）
3. rename orders_temp   → orders                   （新活动表，仅热数据）

create table orders_temp like orders;

-- 分批复制，避免长事务
insert into orders_temp
select * from orders
where create_time >= date_sub(now(), interval 90 day)
   or status not in ('closed', 'cancelled')
   or keep_flag = 1;
-- 按 id 分批 + sleep

-- 应用停写或只读
lock tables orders write;

insert into orders_temp
select * from orders
where (create_time >= ... or status not in (...) or keep_flag = 1)
  and (id > @max_copied_id or updated_at >= @copy_start_time)
on duplicate key update
  user_id = values(user_id),
  amount = values(amount),
  status = values(status),
  updated_at = values(updated_at);

unlock tables;

rename table
  orders to orders_backup_20240818,
  orders_temp to orders;

-- 校正自增，避免新 id 与备份表冲突
set @next_ai = (select max(id) + 1 from orders_backup_20240818);
set @sql = concat('alter table orders auto_increment = ', @next_ai);
prepare stmt from @sql;
execute stmt;
deallocate prepare stmt;

-- 月初：新表已建好 orders_202402
rename table
  orders_202401 to orders_archive_202401,
  orders_202402 to orders;           -- 或应用改指向新表名

归档集合 a = status='closed' and create_time < cutoff
实际搬走   b = a and not 保留规则 r
仍留在线   = a and r

where status = 'closed'
  and create_time < @cutoff
  and keep_flag = 0
  and user_id not in (select user_id from vip_users)  -- 示例
  and amount < 100000

-- ❌ 危险：每批删哪些行不确定，无法对账重跑
delete from orders where ... limit 5000;

-- 应归档且未保留：任务结束后源表应为 0
select count(*) from orders
where status='closed' and create_time < @cutoff and keep_flag = 0;

-- 应保留：仍在源表
select count(*) from orders
where status='closed' and create_time < @cutoff and keep_flag = 1;

-- 主键无交集
select count(*) from orders o inner join orders_archive a on o.id = a.id;

1. 计算本批 source_count、checksum_source
2. insert 归档
3. 计算 archive_count、checksum_archive
4. 不一致 → rollback / 不 delete、告警
5. 一致 → delete 或 commit
6. 写入 job_log

create trigger tr_orders_ins_sync
after insert on orders for each row
begin
  if new.create_time >= date_sub(now(), interval 90 day)
     or new.status not in ('closed','cancelled')
     or new.keep_flag = 1 then
    insert into orders_temp values (...);
  end if;
end;

-- update / delete 同理；切换前 drop trigger，再 rename

-- 低峰执行，大表会锁表或耗时很长
optimize table orders;
-- 或 alter table orders engine=innodb;

create table archive_job_log (
  id bigint auto_increment primary key,
  job_name varchar(64) not null,
  batch_id varchar(32) not null,
  cutoff_time datetime not null,
  id_range_start bigint,
  id_range_end bigint,
  source_count int,
  archive_count int,
  checksum_source varchar(64),
  checksum_archive varchar(64),
  status enum('running','success','failed') not null,
  error_msg text,
  started_at datetime not null,
  finished_at datetime,
  unique key uk_batch (job_name, batch_id)
);

def archive_orders(cutoff: str, batch_size: int = 5000):
    if job_already_running("orders_archive"):
        return

    last_id = 0
    batch_no = 0

    while true:
        batch_no += 1
        batch_id = f"{date.today()}_{batch_no:03d}"

        with db.transaction():
            rows = db.query(
                "select id, ... from orders where ... and id > %s order by id limit %s",
                (last_id, batch_size)
            )
            if not rows:
                break

            ids = [r.id for r in rows]
            checksum_src = checksum(rows)

            db.execute("insert into archive.orders_archive ...", rows)
            checksum_arc = db.query_one(
                "select count(*), sum(...) from archive.orders_archive where batch_id = %s",
                (batch_id,)
            )

            if not verify(len(rows), checksum_src, checksum_arc):
                raise archiveerror("batch verify failed")

            db.execute(
                "delete from orders where id in (%s)",
                (ids,)
            )

            log_success(batch_id, len(rows), checksum_src)
            last_id = max(ids)

        sleep(0.1)  # 降低主库压力

    reconcile_global(cutoff)