本文目录导读：
我来分享几个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)
这些案例覆盖了从简单到复杂的各种去重场景，你可以根据实际需求选择合适的方法。去重策略的选择应该基于数据量、重复率、业务规则和性能要求来综合考虑。
