MySQL大表保留Top X行并删除其余数据的高效实现方案

你的现有方案核心问题在于每次DELETE都要重复执行大结果集的子查询，当rowsToKeep≥10000时，子查询返回的数据集过大，不仅拖慢执行速度，还可能触发MySQL内存限制导致失败。下面是几种更高效的实现方式：

方法一：利用索引定位边界值（最快，优先推荐） #

如果created列没有索引，先创建：CREATE INDEX idx_incident_archive_created ON incident_archive(created DESC);，利用索引快速定位要保留的最后一条记录的边界值，直接删除所有早于该值的记录，避免全表扫描和大子查询。

Python代码示例： #

def do_delete_archive(rowsToKeep):
    if rowsToKeep <= 0:
        return

    db_name = find_environment_value('DB_NAME', False, "dbname")
    db_host = find_environment_value('DB_HOST', False, "host")
    db_user = find_environment_value('DB_USER', False, "username")
    db_pass = find_environment_value('DB_PASS', False, "password")
    db = MySQLdb.connect(host=db_host, user=db_user, passwd=db_pass, db=db_name)
    cursor = db.cursor()

    try:
        # 获取要保留的最后一条记录的created值+最小incident_id，确保同时间的记录不被误删
        cursor.execute("""
            SELECT created, MIN(incident_id) 
            FROM (
                SELECT created, incident_id 
                FROM incident_archive 
                ORDER BY created DESC, incident_id DESC 
                LIMIT %s
            ) AS keep_rows
            ORDER BY created ASC, incident_id ASC 
            LIMIT 1
        """, (rowsToKeep,))
        result = cursor.fetchone()
        if not result:
            print("没有需要保留的记录")
            return
        
        keep_created, keep_min_incident_id = result

        # 批量删除，每次删10000条避免锁表太久
        delete_sql = """
            DELETE FROM incident_archive 
            WHERE created < %s 
               OR (created = %s AND incident_id < %s)
            LIMIT 10000
        """
        rowcount = 1
        while rowcount > 0:
            cursor.execute(delete_sql, (keep_created, keep_created, keep_min_incident_id))
            db.commit()
            rowcount = cursor.rowcount
            print(f"--- 本次删除行数: {rowcount} ---")

    except Exception as e:
        print(f"删除失败: {str(e)}")
        db.rollback()
    finally:
        cursor.close()
        db.close()

方法二：临时表替换法（适合超大规模数据，无锁风险） #

如果表数据量达千万级以上，直接删除会导致长时间锁表影响业务。可以用临时表存储要保留的记录，再原子替换原表：

Python代码示例： #

def do_delete_archive(rowsToKeep):
    if rowsToKeep <= 0:
        return

    db_name = find_environment_value('DB_NAME', False, "dbname")
    db_host = find_environment_value('DB_HOST', False, "host")
    db_user = find_environment_value('DB_USER', False, "username")
    db_pass = find_environment_value('DB_PASS', False, "password")
    db = MySQLdb.connect(host=db_host, user=db_user, passwd=db_pass, db=db_name)
    cursor = db.cursor()

    try:
        # 创建临时表，复制原表结构与索引
        cursor.execute("CREATE TABLE incident_archive_temp LIKE incident_archive;")
        
        # 插入要保留的记录
        cursor.execute("""
            INSERT INTO incident_archive_temp 
            SELECT * FROM incident_archive 
            ORDER BY created DESC, incident_id DESC 
            LIMIT %s
        """, (rowsToKeep,))
        db.commit()
        print(f"已保留 {cursor.rowcount} 条记录到临时表")

        # 原子替换表（几乎无锁，业务中断极短）
        cursor.execute("RENAME TABLE incident_archive TO incident_archive_backup, incident_archive_temp TO incident_archive;")
        db.commit()
        print("表替换完成，原表已重命名为incident_archive_backup")

        # 可选：验证数据无误后删除备份表
        # cursor.execute("DROP TABLE incident_archive_backup;")
        # db.commit()

    except Exception as e:
        print(f"操作失败: {str(e)}")
        db.rollback()
        # 失败时清理临时表
        cursor.execute("DROP TABLE IF EXISTS incident_archive_temp;")
    finally:
        cursor.close()
        db.close()

方法三：优化原DELETE语句（用JOIN替代NOT IN） #

NOT IN在处理大结果集时性能极差，改用LEFT JOIN定位要删除的记录，效率会大幅提升：

Python代码示例： #

def do_delete_archive(rowsToKeep):
    if rowsToKeep <= 0:
        return

    db_name = find_environment_value('DB_NAME', False, "dbname")
    db_host = find_environment_value('DB_HOST', False, "host")
    db_user = find_environment_value('DB_USER', False, "username")
    db_pass = find_environment_value('DB_PASS', False, "password")
    db = MySQLdb.connect(host=db_host, user=db_user, passwd=db_pass, db=db_name)
    cursor = db.cursor()

    try:
        # 每次删除10000条，用JOIN替代NOT IN
        delete_sql = """
            DELETE ia FROM incident_archive ia
            LEFT JOIN (
                SELECT incident_id FROM incident_archive 
                ORDER BY created DESC, incident_id DESC 
                LIMIT %s
            ) keep_rows ON ia.incident_id = keep_rows.incident_id
            WHERE keep_rows.incident_id IS NULL
            LIMIT 10000
        """
        rowcount = 1
        while rowcount > 0:
            cursor.execute(delete_sql, (rowsToKeep,))
            db.commit()
            rowcount = cursor.rowcount
            print(f"--- 本次删除行数: {rowcount} ---")

    except Exception as e:
        print(f"删除失败: {str(e)}")
        db.rollback()
    finally:
        cursor.close()
        db.close()

方案选择建议 #

• 小到中规模数据（百万级）：优先用方法一，利用索引速度最快，对业务影响小
• 超大规模数据（千万级以上）：用方法二，避免长时间锁表，保证业务可用性
• 不想大幅修改原有代码：用方法三，替换NOT IN为JOIN，解决rowsToKeep过大的问题

内容的提问来源于stack exchange，提问作者DeadlyDagger