搞定删除重复数据保留一条：从报错到源码解析的实战指南

配置环境就卡半天，是不是熟悉的感觉？跑个脚本删个重数据，结果环境依赖打架，SQL 语法报错，或者数据删了但主键冲突，那种抓狂感真让人想砸键盘。别急，今天咱们不整虚的，直接上手一个完整的实战项目，通过源码解析彻底搞懂删除重复数据保留一条的底层逻辑。

项目目标与痛点场景

在真实的生产环境中，数据重复是常态。无论是用户注册时的并发提交，还是数据迁移时的脚本失误，重复数据（Duplicate Data）都会导致统计偏差、业务逻辑错误甚至系统崩溃。

我们的目标很明确：构建一个可复现的 Python 工具，能够安全、高效地删除重复数据保留一条记录。这里的“保留一条”，指的是保留最新（MAX(id)）或最旧（MIN(id)）的那一条，其余标记删除或物理删除。

为什么需要源码解析？因为网上大多数教程只给 DELETE 语句，一旦遇到大表、高并发或者特定数据库（如 MySQL、PostgreSQL、Oracle）的差异，代码直接崩盘。我们需要深入到底层执行逻辑，理解事务隔离级别、索引选择以及锁机制，才能写出真正健壮的代码。

目录结构与依赖环境

为了避免“配置环境就卡半天”的噩梦，我们先定义一个最小化、可复现的项目结构。不要使用复杂的框架，就用最原生的 Python 标准库和数据库驱动。

project_structure/
├── requirements.txt
├── config.py          # 数据库配置
├── db_utils.py        # 数据库连接池与执行器
├── duplicate_cleaner.py # 核心清洗逻辑
├── test_data.py       # 生成测试数据
└── main.py            # 入口脚本

requirements.txt 内容如下，保持精简：

pymysql==1.0.2
psycopg2-binary==2.9.5
loguru==0.7.0

关键避坑点：很多新人喜欢用 SQLAlchemy 做 ORM，但在处理大规模数据清洗时，ORM 的抽象层会带来巨大的内存开销和性能损耗。对于删除重复数据保留一条这种底层操作，直接使用 SQL 或 DB-API 接口是更高效、更可控的选择。

config.py 中我们定义多数据库支持，通过配置切换：

import os# 环境变量覆盖默认配置，方便 CI/CD 部署
DB_CONFIG = {"engine": os.getenv("DB_ENGINE", "mysql"),  # mysql, postgres"host": os.getenv("DB_HOST", "127.0.0.1"),"port": int(os.getenv("DB_PORT", 3306)),"user": os.getenv("DB_USER", "root"),"password": os.getenv("DB_PASSWORD", "password"),"database": os.getenv("DB_NAME", "test_db"),"batch_size": int(os.getenv("BATCH_SIZE", 1000))  # 批量处理大小
}

核心代码实现与源码解析

这是本文的核心部分。我们将分两步走：先写一个“能跑”的版本，再通过源码解析找出它的隐患，最后给出优化后的生产级代码。

1. 基础实现：朴素 SQL 的陷阱

很多博客教你这样写：

DELETE t1 FROM users t1
INNER JOIN users t2
WHERE t1.id > t2.id
AND t1.email = t2.email;

这招在 MySQL 小表上确实好用。但在 Postgres 里，语法完全不同；在 Oracle 里，DELETE 不能直接子查询同一张表。更致命的是，这种写法没有事务控制，没有进度反馈，一旦数据量上百万，连接超时直接中断，数据处于“半删除”状态，灾难就此发生。

2. 生产级实现：Python 封装与批量处理

我们使用 loguru 记录日志，使用批量处理（Batching）避免长事务。

# db_utils.py
import pymysql
import psycopg2
from contextlib import contextmanager
from loguru import logger
import config@contextmanager
def get_db_cursor():"""上下文管理器，确保连接和游标正确关闭"""conn = Nonecursor = Nonetry:if config.DB_CONFIG["engine"] == "mysql":conn = pymysql.connect(host=config.DB_CONFIG["host"],port=config.DB_CONFIG["port"],user=config.DB_CONFIG["user"],password=config.DB_CONFIG["password"],database=config.DB_CONFIG["database"],charset='utf8mb4',autocommit=False  # 关键：手动提交，控制事务)elif config.DB_CONFIG["engine"] == "postgres":conn = psycopg2.connect(host=config.DB_CONFIG["host"],port=config.DB_CONFIG["port"],user=config.DB_CONFIG["user"],password=config.DB_CONFIG["password"],dbname=config.DB_CONFIG["database"],)else:raise ValueError(f"Unsupported engine: {config.DB_CONFIG['engine']}")cursor = conn.cursor()yield cursorexcept Exception as e:logger.error(f"Database error: {e}")if conn:conn.rollback()raisefinally:if cursor:cursor.close()if conn:conn.close()

接下来是核心清洗逻辑 duplicate_cleaner.py。这里我们采用“查找重复组 -> 保留最大 ID -> 删除其余”的策略。

# duplicate_cleaner.py
import config
from db_utils import get_db_cursor
from loguru import loggerdef find_duplicate_groups(table_name: str, unique_columns: list):"""查找所有存在重复的分组返回: [(group_values, max_id, count), ...]"""columns_str = ", ".join(unique_columns)# 注意：这里使用 GROUP BY 和 HAVING，避免全表扫描sql = f"""SELECT {columns_str}, MAX(id) as max_id, COUNT(*) as cntFROM {table_name}GROUP BY {columns_str}HAVING COUNT(*) > 1"""results = []with get_db_cursor() as cursor:cursor.execute(sql)for row in cursor.fetchall():# row 结构: (col1, col2, ..., max_id, count)group_vals = row[:-2]max_id = row[-2]cnt = row[-1]results.append((group_vals, max_id, cnt))return resultsdef delete_duplicates(table_name: str, unique_columns: list, batch_size: int = 1000):"""执行删除操作，保留每组中 ID 最大的一条"""# 1. 获取重复组dup_groups = find_duplicate_groups(table_name, unique_columns)if not dup_groups:logger.info("No duplicates found.")return 0logger.info(f"Found {len(dup_groups)} duplicate groups. Starting cleanup...")total_deleted = 0columns_str = ", ".join(unique_columns)# 2. 分批处理，避免长事务for i in range(0, len(dup_groups), batch_size):batch = dup_groups[i:i + batch_size]with get_db_cursor() as cursor:# 构造 DELETE 语句# 对于 MySQL，可以直接使用 NOT IN 或子查询# 对于 Postgres，需要使用 CTE 或 JOINif config.DB_CONFIG["engine"] == "mysql":# 利用主键索引加速删除# 注意：这里假设 id 是主键delete_sql = f"""DELETE FROM {table_name}WHERE id NOT IN (SELECT max_id FROM (SELECT MAX(id) as max_idFROM {table_name}WHERE ({columns_str} = %s, %s, ...)) AS tmp)"""# 简化逻辑：直接针对每个 group 执行删除# 为了演示通用性，这里使用更稳妥的逐个 Group 处理逻辑for group_vals, max_id, cnt in batch:placeholders = ", ".join(["%s"] * len(group_vals))# 删除该组中 ID 不等于 max_id 的记录del_sql = f"""DELETE FROM {table_name}WHERE ({columns_str}) IN (({placeholders}))AND id != %s"""cursor.execute(del_sql, (*group_vals, max_id))total_deleted += cursor.rowcountelif config.DB_CONFIG["engine"] == "postgres":# Postgres 语法差异for group_vals, max_id, cnt in batch:placeholders = ", ".join(["%s"] * len(group_vals))del_sql = f"""DELETE FROM {table_name}WHERE ({columns_str}) IN (({placeholders}))AND id != %s"""cursor.execute(del_sql, (*group_vals, max_id))total_deleted += cursor.rowcount# 每批提交一次# MySQL 和 PG 都需要显式提交if config.DB_CONFIG["engine"] == "mysql":cursor.connection.commit()else:cursor.connection.commit()logger.info(f"Processed batch {i//batch_size + 1}. Deleted {cursor.rowcount} rows in last batch.")logger.info(f"Cleanup complete. Total deleted: {total_deleted}")return total_deleted

源码解析关键点：

为什么分批？ 如果一次删除 10 万条，事务日志（Redo Log/WAL）会激增，可能撑爆磁盘或导致主从延迟。分批提交可以将长事务拆分为短事务，降低锁持有时间。
为什么保留 MAX(id)？ 在自增 ID 场景中，ID 越大通常代表数据越新。保留最大 ID 是最常见的业务需求。如果业务需要保留最小 ID，只需将 MAX 改为 MIN。
事务隔离级别：我们在 db_utils 中禁用了 autocommit。这意味着在 commit() 之前，所有 DELETE 操作都在一个事务中。如果中途出错，rollback() 会回滚所有操作，保证数据一致性。

运行与测试：复现真实场景

光看代码不够，得跑起来。我们编写 test_data.py 生成脏数据。

# test_data.py
from db_utils import get_db_cursor
import random
import stringdef create_test_table():sql = """CREATE TABLE IF NOT EXISTS users (id INT AUTO_INCREMENT PRIMARY KEY,email VARCHAR(255) NOT NULL,name VARCHAR(100),created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP);"""# Postgres 语法不同，这里仅展示 MySQL 逻辑，PG 需适配 SERIAL 类型with get_db_cursor() as cursor:cursor.execute("DROP TABLE IF EXISTS users")cursor.execute(sql)cursor.connection.commit()def insert_test_data(count=10000):with get_db_cursor() as cursor:for i in range(count):# 故意制造重复：每 10 条中随机选一个重复邮箱base_email = f"user{random.randint(0, 900)}@test.com"name = ''.join(random.choices(string.ascii_uppercase, k=5))cursor.execute("INSERT INTO users (email, name) VALUES (%s, %s)",(base_email, name))cursor.connection.commit()print(f"Inserted {count} rows with intentional duplicates.")if __name__ == "__main__":create_test_table()insert_test_data()

运行 main.py：

# main.py
from duplicate_cleaner import delete_duplicatesif __name__ == "__main__":table = "users"# 假设 email 是业务唯一键，但数据库层面没有唯一约束unique_cols = ["email"]deleted_count = delete_duplicates(table, unique_cols, batch_size=500)print(f"Total records removed: {deleted_count}")

测试验证：运行后，检查 users 表：

SELECT email, COUNT(*) from users GROUP BY email HAVING COUNT(*) > 1;

结果应为空。同时，抽查某个邮箱，确认保留的是 ID 最大的那条记录。

优化扩展与避坑指南

在 Stack Overflow 上，关于“删除重复数据”的问题高达数千个。总结下来，常见的坑有这三个：

索引缺失：如果 email 字段没有索引，GROUP BY 会导致全表扫描，大表直接卡死。对策：在执行清洗前，确保 unique_columns 上有组合索引。
```
CREATE INDEX idx_email ON users(email);
```
外键约束：如果 id 被其他表引用，直接 DELETE 会报错。对策：在清洗前，检查外键依赖，或使用 ON DELETE CASCADE（慎用），或者先软删除（标记 is_deleted=1），再异步物理删除。
并发写入：清洗过程中，如果有新数据插入相同的重复值，会导致刚删完又出现重复。对策：在清洗期间，暂停写入服务，或在应用层加分布式锁。对于高并发场景，建议先 UPDATE 标记，再异步清理，最后 DELETE 标记数据，实现无感清洗。

进阶技巧：使用视图辅助 如果重复逻辑非常复杂，可以创建一个临时视图来标识重复项：

CREATE TEMPORARY VIEW dup_users AS
SELECT id, email, ROW_NUMBER() OVER (PARTITION BY email ORDER BY id DESC) as rn
FROM users;DELETE FROM users
WHERE id IN (SELECT id FROM dup_users WHERE rn > 1);

这种写法在 PostgreSQL 中非常优雅，利用了窗口函数 ROW_NUMBER()，代码可读性远优于嵌套子查询。