Python数据库去重案例如何筛选唯一数据

wen python案例 2026-06-30 258

本文目录导读：

基础SQL去重 - SQLite案例
使用Pandas高级去重
复杂场景：多条件去重
实时数据去重（流式处理）
大数据量去重优化
实际应用：电商订单去重
总结与建议

我来分享几个Python数据库去重的实用案例,涵盖不同场景和实现方式。

基础SQL去重 - SQLite案例

import sqlite3
# 创建测试数据
conn = sqlite3.connect('test.db')
cursor = conn.cursor()
# 创建表并插入重复数据
cursor.execute('''
    CREATE TABLE IF NOT EXISTS users (
        id INTEGER PRIMARY KEY,
        name TEXT,
        email TEXT,
        age INTEGER
    )
''')
# 插入重复数据
test_data = [
    ('张三', 'zhangsan@email.com', 25),
    ('李四', 'lisi@email.com', 30),
    ('张三', 'zhangsan@email.com', 25),  # 重复
    ('王五', 'wangwu@email.com', 28),
    ('张三', 'zhangsan@email.com', 25),  # 重复
]
cursor.executemany('INSERT INTO users (name, email, age) VALUES (?, ?, ?)', test_data)
conn.commit()
# 方法1：使用DISTINCT去重
print("=== DISTINCT去重 ===")
cursor.execute('SELECT DISTINCT name, email, age FROM users')
for row in cursor.fetchall():
    print(f"姓名: {row[0]}, 邮箱: {row[1]}, 年龄: {row[2]}")
# 方法2：使用GROUP BY去重
print("\n=== GROUP BY去重 ===")
cursor.execute('''
    SELECT name, email, age, COUNT(*) as count 
    FROM users 
    GROUP BY name, email, age
    HAVING COUNT(*) > 0
''')
for row in cursor.fetchall():
    print(f"姓名: {row[0]}, 邮箱: {row[1]}, 年龄: {row[2]}, 重复次数: {row[3]}")
conn.close()

使用Pandas高级去重

import pandas as pd
from datetime import datetime
# 创建包含重复数据的DataFrame
data = {
    '订单号': ['A001', 'A002', 'A001', 'A003', 'A002', 'A004'],
    '商品': ['苹果', '香蕉', '苹果', '橘子', '香蕉', '葡萄'],
    '数量': [5, 3, 5, 8, 4, 6],
    '日期': ['2024-01-01', '2024-01-02', '2024-01-01', '2024-01-03', '2024-01-02', '2024-01-04']
}
df = pd.DataFrame(data)
print("=== 原始数据 ===")
print(df)
# 方法1：完全重复行去重
df_unique = df.drop_duplicates()
print("\n=== 完全去重 ===")
print(df_unique)
# 方法2：指定列去重（保留第一次出现的）
df_partial = df.drop_duplicates(subset=['订单号'], keep='first')
print("\n=== 按订单号去重（保留第一个）===")
print(df_partial)
# 方法3：指定列去重（保留最后一次出现的）
df_last = df.drop_duplicates(subset=['订单号'], keep='last')
print("\n=== 按订单号去重（保留最后一个）===")
print(df_last)
# 方法4：基于条件的去重（保留数量最大的）
df_max = df.loc[df.groupby('订单号')['数量'].idxmax()]
print("\n=== 按订单号去重（保留数量最大的）===")
print(df_max)

复杂场景：多条件去重

import pandas as pd
from datetime import datetime, timedelta
import random
# 模拟销售数据
def generate_sales_data():
    products = ['手机', '电脑', '平板', '耳机', '手表']
    cities = ['北京', '上海', '广州', '深圳', '杭州']
    data = []
    # 创建一些重复数据
    repeat_data = [
        ('手机', '北京', 5000, '2024-01-01'),
        ('手机', '北京', 5000, '2024-01-01'),  # 重复
        ('电脑', '上海', 8000, '2024-01-02'),
        ('电脑', '上海', 8000, '2024-01-02'),  # 重复
        ('平板', '广州', 3000, '2024-01-03'),
    ]
    for product, city, price, date in repeat_data:
        data.append({'产品': product, '城市': city, '价格': price, '日期': date})
    # 添加一些不重复的数据
    for i in range(10):
        data.append({
            '产品': random.choice(products),
            '城市': random.choice(cities),
            '价格': random.randint(1000, 10000),
            '日期': f'2024-01-{(i+10):02d}'
        })
    return pd.DataFrame(data)
df = generate_sales_data()
print("=== 原始销售数据 ===")
print(df)
# 多条件去重：产品和城市组合唯一
df_multi_unique = df.drop_duplicates(subset=['产品', '城市'], keep='first')
print("\n=== 产品和城市组合去重 ===")
print(df_multi_unique)
# 按日期范围去重（最近一周的数据去重）
df['日期'] = pd.to_datetime(df['日期'])
last_week = df[df['日期'] >= (df['日期'].max() - timedelta(days=7))]
print("\n=== 最近一周数据去重 ===")
print(last_week.drop_duplicates(subset=['产品', '城市']))

实时数据去重（流式处理）

from collections import defaultdict
import time
import hashlib
class DeduplicationHandler:
    """实时数据去重处理器"""
    def __init__(self, window_size=1000):
        self.window_size = window_size
        self.data_window = []
        self.hash_set = set()
    def _create_hash(self, data):
        """创建数据指纹"""
        # 将数据转换为字符串并生成哈希
        data_str = str(sorted(data.items()))
        return hashlib.md5(data_str.encode()).hexdigest()
    def process_data(self, data):
        """处理新数据，返回False表示重复"""
        data_hash = self._create_hash(data)
        if data_hash in self.hash_set:
            return False  # 重复数据
        # 添加到窗口
        self.data_window.append((data_hash, data))
        self.hash_set.add(data_hash)
        # 维护窗口大小
        if len(self.data_window) > self.window_size:
            old_hash, old_data = self.data_window.pop(0)
            self.hash_set.remove(old_hash)
        return True  # 新数据
    def get_unique_count(self):
        return len(self.data_window)
# 使用示例
def simulate_realtime_data():
    dedup = DeduplicationHandler(window_size=5)
    test_data = [
        {'sensor': 'A', 'value': 25.5, 'time': '10:00:01'},
        {'sensor': 'B', 'value': 30.0, 'time': '10:00:02'},
        {'sensor': 'A', 'value': 25.5, 'time': '10:00:01'},  # 重复
        {'sensor': 'C', 'value': 28.0, 'time': '10:00:03'},
        {'sensor': 'A', 'value': 25.5, 'time': '10:00:01'},  # 重复
    ]
    print("=== 实时数据去重测试 ===")
    for i, data in enumerate(test_data):
        result = dedup.process_data(data)
        status = "✓ 新数据" if result else "✗ 重复数据"
        print(f"数据{i+1}: {data} -> {status}")
    print(f"\n去重后唯一数据数: {dedup.get_unique_count()}")
simulate_realtime_data()

大数据量去重优化

import numpy as np
import pandas as pd
from concurrent.futures import ProcessPoolExecutor
import time
def optimized_dedup_large_dataset():
    """大数据量去重优化示例"""
    # 生成100万条数据
    np.random.seed(42)
    print("生成100万条测试数据...")
    n_rows = 1000000
    df = pd.DataFrame({
        'id': np.random.randint(0, 500000, n_rows),
        'category': np.random.choice(['A', 'B', 'C', 'D'], n_rows),
        'value': np.random.randn(n_rows),
        'timestamp': pd.date_range('2024-01-01', periods=n_rows, freq='1min')
    })
    print(f"原始数据大小: {df.shape}")
    # 方法1：使用索引去重（最快）
    start_time = time.time()
    df_index_dedup = df.set_index('id').index.unique()
    method1_time = time.time() - start_time
    print(f"\n方法1（索引去重）: {len(df_index_dedup)} 条数据, 耗时: {method1_time:.3f}秒")
    # 方法2：使用groupby去重
    start_time = time.time()
    df_groupby_dedup = df.drop_duplicates(subset=['id', 'category'])
    method2_time = time.time() - start_time
    print(f"方法2（groupby去重）: {len(df_groupby_dedup)} 条数据, 耗时: {method2_time:.3f}秒")
    # 方法3：使用哈希去重（适合大集合）
    start_time = time.time()
    # 创建哈希集合
    hash_set = set()
    for idx, row in df.iterrows():
        hash_val = hash((row['id'], row['category']))
        hash_set.add(hash_val)
    method3_time = time.time() - start_time
    print(f"方法3（哈希去重）: {len(hash_set)} 个哈希值, 耗时: {method3_time:.3f}秒")
# 注意：在大数据量时，建议使用前两种方法
# optimized_dedup_large_dataset()
print("大数据量去重示例已准备，取消注释运行")

实际应用：电商订单去重

import pandas as pd
from datetime import datetime, timedelta
import uuid
class OrderDeduplication:
    """电商订单去重系统"""
    def __init__(self):
        self.processed_orders = set()
        self.duplicate_log = []
    def process_orders(self, orders_df):
        """
        处理订单去重
        去重规则：
        1. 同一用户同一商品在24小时内下多次订单视为重复
        2. 相同的收货地址和联系电话视为重复
        """
        results = []
        for idx, order in orders_df.iterrows():
            order_key = self._create_order_key(order)
            if self._is_unique(order, order_key):
                results.append(order)
                self.processed_orders.add(order_key)
            else:
                self.duplicate_log.append({
                    'order_id': order['order_id'],
                    'reason': '重复订单',
                    'timestamp': datetime.now()
                })
        return pd.DataFrame(results)
    def _create_order_key(self, order):
        """创建订单唯一标识"""
        # 组合用户ID、商品ID、收货地址和联系电话
        key_parts = [
            str(order.get('user_id', '')),
            str(order.get('product_id', '')),
            str(order.get('address', '')),
            str(order.get('phone', ''))
        ]
        return '_'.join(key_parts)
    def _is_unique(self, order, order_key):
        """检查订单是否唯一"""
        if order_key in self.processed_orders:
            return False
        return True
# 模拟订单数据
def simulate_order_dedup():
    dedup = OrderDeduplication()
    # 创建测试订单数据
    orders_data = {
        'order_id': [str(uuid.uuid4()) for _ in range(10)],
        'user_id': [1001, 1002, 1001, 1003, 1002, 1001, 1004, 1005, 1001, 1006],
        'product_id': ['P001', 'P002', 'P001', 'P003', 'P002', 'P004', 'P005', 'P006', 'P001', 'P007'],
        'address': ['北京路1号', '上海路2号', '北京路1号', '广州路3号', 
                    '深圳路4号', '北京路1号', '杭州路5号', '成都路6号', 
                    '北京路1号', '武汉路7号'],
        'phone': ['13800138001', '13800138002', '13800138001', '13800138003',
                  '13800138002', '13800138001', '13800138004', '13800138005',
                  '13800138001', '13800138006'],
        'amount': [100, 200, 100, 300, 200, 150, 400, 500, 100, 600]
    }
    orders_df = pd.DataFrame(orders_data)
    print("=== 原始订单数据 ===")
    print(orders_df[['order_id', 'user_id', 'product_id', 'address', 'amount']])
    # 执行去重
    unique_orders = dedup.process_orders(orders_df)
    print(f"\n=== 去重后的订单 ({len(unique_orders)} 条) ===")
    print(unique_orders[['user_id', 'product_id', 'address', 'amount']])
    print(f"\n被标记为重复的订单数: {len(dedup.duplicate_log)}")
    for dup in dedup.duplicate_log:
        print(f"订单 {dup['order_id']}: {dup['reason']}")
simulate_order_dedup()

总结与建议

选择合适的去重方法

小数据集（<1万条）：使用drop_duplicates()或SQL的DISTINCT
中等数据集（1-100万条）：使用Pandas的groupby或索引去重
大数据集（>100万条）：考虑分布式处理或数据库层面的去重

性能优化技巧

# 1. 使用索引加速
df.set_index('id', inplace=True)
df = df[~df.index.duplicated(keep='first')]
# 2. 使用内存映射（大文件）
df = pd.read_csv('large_file.csv', usecols=['col1', 'col2'], dtype={'col1': 'category'})
# 3. 分批处理
def batch_dedup(df, batch_size=10000):
    results = []
    for i in range(0, len(df), batch_size):
        batch = df.iloc[i:i+batch_size]
        deduped = batch.drop_duplicates(subset=['key'])
        results.append(deduped)
    return pd.concat(results).drop_duplicates(subset=['key'])
# 4. 使用bloom filter（适合大规模去重）
from pybloom_live import BloomFilter
bloom = BloomFilter(capacity=1000000, error_rate=0.001)

这些案例覆盖了从简单到复杂的各种去重场景，你可以根据实际需求选择合适的方法。去重策略的选择应该基于数据量、重复率、业务规则和性能要求来综合考虑。

本文地址： https://www.bmjup.com/post/12262.html