MySQL - 查找重复记录

MySQL - 如何查找和管理重复记录

数据库中的重复记录可能导致数据完整性问题、不准确的分析结果以及性能下降。识别和管理这些重复数据是数据库维护中的一项关键任务。本教程将探讨在 MySQL 中查找和处理重复数据的现代且有效的方法。

虽然最佳策略是使用 PRIMARY KEY（主键）和 UNIQUE（唯一）等约束来阻止重复数据的创建，但重复数据仍然可能由于应用程序错误、数据导入或遗留系统迁移而出现。我们将介绍两种主要的检测技术：传统的 GROUP BY 与 HAVING 子句结合使用，以及更强大的窗口函数 ROW_NUMBER()。

场景设置

我们以 products 表为例。请注意，有些产品具有相同的名称和类别，在我们的示例中，我们将这些视为重复数据。

CREATE TABLE products (
    id INT AUTO_INCREMENT PRIMARY KEY,
    product_name VARCHAR(100) NOT NULL,
    category VARCHAR(50) NOT NULL,
    price DECIMAL(10, 2),
    stock_quantity INT
);

INSERT INTO products (product_name, category, price, stock_quantity) VALUES
('Laptop', 'Electronics', 1200.00, 50),
('Keyboard', 'Electronics', 75.00, 200),
('Laptop', 'Electronics', 1250.00, 35), -- 重复的名称和类别
('Mouse', 'Electronics', 25.00, 500),
('T-Shirt', 'Apparel', 20.00, 300),
('T-Shirt', 'Apparel', 22.50, 250),   -- 重复的名称和类别
('Jeans', 'Apparel', 60.00, 150);

方法 1：使用 GROUP BY 和 HAVING

这是查找哪些值重复的经典方法。它的工作原理是根据您怀疑存在重复数据的列对行进行分组，然后使用 HAVING 子句筛选出计数大于 1 的组。

概念

• GROUP BY 子句根据指定列中相同的值将行聚合成汇总行。
• COUNT() 聚合函数计算每个组中的行数。
• HAVING 子句筛选这些组，只保留计数大于 1 的组。

示例：按 product_name 和 category 查找重复数据

此查询识别 product_name 和 category 的哪些组合出现了多次，并显示它们的计数。

SELECT
    product_name,
    category,
    COUNT(*) AS occurrences
FROM
    products
GROUP BY
    product_name, category
HAVING
    COUNT(*) > 1;

输出

产品名称	类别	出现次数
Laptop	Electronics	2
T-Shirt	Apparel	2

局限性： 这种方法非常适合识别哪些值重复，但它不容易返回所有单个重复行的完整详细信息（如 id、price 等）。

方法 2：使用窗口函数（ROW_NUMBER）

从 MySQL 8.0 开始，窗口函数提供了一种更强大、更灵活的方式来处理重复数据。ROW_NUMBER() 函数可以为分区（行组）内的每一行分配一个唯一的排名，从而让您轻松识别和选择重复数据。

概念

• PARTITION BY 子句根据指定列（例如 product_name、category）将行划分为分区（组）。这类似于 GROUP BY，但不会折叠行。
• ROW_NUMBER() 函数为其分区内的每一行分配一个顺序整数。
• 任何被分配了大于 1 的 ROW_NUMBER() 的行都是其分区内的重复数据。

示例：查找并列出所有重复行

我们使用公共表表达式（Common Table Expression, CTE）首先对行进行编号。然后，我们从 CTE 中选择，以查找所有编号大于 1 的行。

WITH NumberedProducts AS (
    SELECT
        *,
        ROW_NUMBER() OVER (PARTITION BY product_name, category ORDER BY id) as row_num
    FROM
        products
)
SELECT id, product_name, category, price, stock_quantity
FROM NumberedProducts
WHERE row_num > 1;

输出

此查询返回重复条目的完整记录，使其易于检查或删除。

ID	产品名称	类别	价格	库存数量
3	Laptop	Electronics	1250.00	35
6	T-Shirt	Apparel	22.50	250

实际应用：删除重复数据

ROW_NUMBER() 方法在删除重复数据时特别有用，同时保留其中一个版本的记录（通常是第一个，基于 OVER 子句中的 ORDER BY）。

-- 注意：在运行 DELETE 语句之前，请务必备份您的数据。
-- 您可以先运行 SELECT 部分来测试将要删除的内容。

DELETE p
FROM products AS p
INNER JOIN (
    SELECT
        id,
        ROW_NUMBER() OVER (PARTITION BY product_name, category ORDER BY id) as row_num
    FROM
        products
) AS numbered_products ON p.id = numbered_products.id
WHERE numbered_products.row_num > 1;

最佳实践与性能

• 预防胜于治疗： 在不应有重复值的列上使用 UNIQUE（唯一）约束，从源头防止问题。
• 索引： 确保 GROUP BY 或 PARTITION BY 子句中使用的列已建立索引。这可以显著提高大型表上查找重复数据的查询性能。
• 选择正确的方法： 使用 GROUP BY 进行重复值的快速汇总。当您需要检查、更新或删除完整的重复行时，使用 ROW_NUMBER()。

使用客户端程序查找重复数据

您也可以从应用程序代码中执行这些查询，以编程方式查找重复数据。以下是使用现代最佳实践的示例。

示例代码

Python
NodeJS
Java
PHP



import mysql.connector
from mysql.connector import errorcode

# 最佳实践：使用字典传递连接参数
config = {
    'user': 'your_user',
    'password': 'your_password',
    'host': '127.0.0.1',
    'database': 'your_database'
}

query = """
    SELECT
        product_name, category, COUNT(*) AS occurrences
    FROM products
    GROUP BY product_name, category
    HAVING COUNT(*) > 1
"""

try:
    # 使用 'with' 语句进行资源自动管理
    with mysql.connector.connect(**config) as connection:
        with connection.cursor(dictionary=True) as cursor:
            cursor.execute(query)
            duplicates = cursor.fetchall()

            if duplicates:
                print("Found duplicate records:")
                for row in duplicates:
                    print(f"- Product: {row['product_name']}, Category: {row['category']}, Occurrences: {row['occurrences']}")
            else:
                print("No duplicate records found.")

except mysql.connector.Error as err:
    if err.errno == errorcode.ER_ACCESS_DENIED_ERROR:
        print("Authentication error")
    elif err.errno == errorcode.ER_BAD_DB_ERROR:
        print("Database does not exist")
    else:
        print(err)



const mysql = require('mysql2/promise');

// 使用连接池以获得更好的性能和资源管理
const pool = mysql.createPool({
    host: '127.0.0.1',
    user: 'your_user',
    password: 'your_password',
    database: 'your_database',
    waitForConnections: true,
    connectionLimit: 10,
    queueLimit: 0
});

const findDuplicates = async () => {
    const query = `
        SELECT
            product_name, category, COUNT(*) AS occurrences
        FROM products
        GROUP BY product_name, category
        HAVING COUNT(*) > 1
    `;

    let connection;
    try {
        connection = await pool.getConnection();
        const [rows] = await connection.execute(query);

        if (rows.length > 0) {
            console.log('Found duplicate records:');
            rows.forEach(row => {
                console.log(`- Product: ${row.product_name}, Category: ${row.category}, Occurrences: ${row.occurrences}`);
            });
        } else {
            console.log('No duplicate records found.');
        }
    } catch (error) {
        console.error('An error occurred while finding duplicates:', error);
    } finally {
        if (connection) connection.release(); // 将连接释放回连接池
        await pool.end(); // 完成后关闭连接池
    }
};

findDuplicates();



import java.sql.*;

public class FindDuplicates {
    // 为连接详细信息使用常量
    private static final String DB_URL = "jdbc:mysql://localhost:3306/your_database";
    private static final String USER = "your_user";
    private static final String PASS = "your_password";

    public static void main(String[] args) {
        String query = """
            SELECT
                product_name, category, COUNT(*) AS occurrences
            FROM products
            GROUP BY product_name, category
            HAVING COUNT(*) > 1
        """

        // 使用 try-with-resources 进行资源自动管理
        try (Connection conn = DriverManager.getConnection(DB_URL, USER, PASS);
             Statement stmt = conn.createStatement();
             ResultSet rs = stmt.executeQuery(query)) {

            boolean found = false;
            while (rs.next()) {
                if (!found) {
                    System.out.println("Found duplicate records:");
                    found = true;
                }
                String productName = rs.getString("product_name");
                String category = rs.getString("category");
                int occurrences = rs.getInt("occurrences");
                System.out.printf("- Product: %s, Category: %s, Occurrences: %d%n", productName, category, occurrences);
            }

            if (!found) {
                System.out.println("No duplicate records found.");
            }

        } catch (SQLException e) {
            // 基本错误处理
            e.printStackTrace();
        }
    }
}



<?php
// 数据库配置
$host = '127.0.0.1';
$db   = 'your_database';
$user = 'your_user';
$pass = 'your_password';
$charset = 'utf8mb4';

// 设置 DSN (数据源名称)
$dsn = "mysql:host=$host;dbname=$db;charset=$charset";
$options = [
    PDO::ATTR_ERRMODE            => PDO::ERRMODE_EXCEPTION,
    PDO::ATTR_DEFAULT_FETCH_MODE => PDO::FETCH_ASSOC,
    PDO::ATTR_EMULATE_PREPARES   => false,
];

$query = "
    SELECT
        product_name, category, COUNT(*) AS occurrences
    FROM products
    GROUP BY product_name, category
    HAVING COUNT(*) > 1
";

try {
    // 使用 PDO 进行数据库连接
    $pdo = new PDO($dsn, $user, $pass, $options);

    $stmt = $pdo->query($query);
    $duplicates = $stmt->fetchAll();

    if ($duplicates) {
        echo "Found duplicate records:\n";
        foreach ($duplicates as $row) {
            echo "- Product: {$row['product_name']}, Category: {$row['category']}, Occurrences: {$row['occurrences']}\n";
        }
    } else {
        echo "No duplicate records found.\n";
    }

} catch (\PDOException $e) {
    // 在实际应用中，您应该记录此错误，而不是将其显示给用户。
    throw new \PDOException($e->getMessage(), (int)$e->getCode());
}

?>