本文目录导读：
我来介绍几种Python去重整合缓存重复数据的方法：
基于字典的去重（保留最后出现的数据）
def dedup_by_dict(data_list, key_func=None):
"""
使用字典去重，保留最后出现的数据
:param data_list: 原始数据列表
:param key_func: 提取键的函数，默认使用元素本身
:return: 去重后的列表
"""
if key_func is None:
key_func = lambda x: x
seen = {}
for item in data_list:
key = key_func(item)
seen[key] = item # 后面的会覆盖前面的
return list(seen.values())
# 示例使用
cache_data = [
{"id": 1, "name": "Alice"},
{"id": 2, "name": "Bob"},
{"id": 1, "name": "Alice Updated"}
]
result = dedup_by_dict(cache_data, key_func=lambda x: x["id"])
print(result)
# 输出: [{"id": 1, "name": "Alice Updated"}, {"id": 2, "name": "Bob"}]
基于集合的去重（保留首次出现的数据）
def dedup_by_set(data_list, key_func=None):
"""
使用集合去重，保留首次出现的数据
:param data_list: 原始数据列表
:param key_func: 提取键的函数
:return: 去重后的列表
"""
if key_func is None:
key_func = lambda x: x
seen = set()
result = []
for item in data_list:
key = key_func(item)
if key not in seen:
seen.add(key)
result.append(item)
return result
# 示例使用
cache_data = [1, 2, 3, 2, 1, 4, 5, 3]
result = dedup_by_set(cache_data)
print(result) # 输出: [1, 2, 3, 4, 5]
使用OrderedDict保持插入顺序
from collections import OrderedDict
def dedup_with_order(data_list, key_func=None):
"""
使用OrderedDict去重，保持原始顺序
"""
if key_func is None:
key_func = lambda x: x
# OrderedDict会保留首次插入的顺序
seen = OrderedDict()
for item in data_list:
key = key_func(item)
if key not in seen: # 只保留第一次出现的
seen[key] = item
return list(seen.values())
缓存数据整合去重（复杂对象）
from datetime import datetime
import json
class CacheDataMerger:
def __init__(self):
self.cache = {}
def add_data(self, key, value, timestamp=None):
"""
添加数据到缓存，自动去重并保留最新数据
"""
if timestamp is None:
timestamp = datetime.now()
if key in self.cache:
existing_time = self.cache[key].get('timestamp')
if existing_time and timestamp > existing_time:
# 新数据更新时间更晚，进行更新
self.cache[key] = {
'value': value,
'timestamp': timestamp,
'version': self.cache[key].get('version', 1) + 1
}
else:
self.cache[key] = {
'value': value,
'timestamp': timestamp,
'version': 1
}
def get_all_data(self):
"""获取所有缓存数据"""
return {
key: data['value']
for key, data in self.cache.items()
}
def merge_from_other(self, other_cache):
"""合并另一个缓存数据"""
for key, data in other_cache.items():
self.add_data(key, data['value'], data.get('timestamp'))
# 示例使用
merger = CacheDataMerger()
merger.add_data("user_1", {"name": "Alice"}, datetime(2024, 1, 1))
merger.add_data("user_2", {"name": "Bob"}, datetime(2024, 1, 2))
merger.add_data("user_1", {"name": "Alice Updated"}, datetime(2024, 1, 3))
print(merger.get_all_data())
# 输出: {'user_1': {'name': 'Alice Updated'}, 'user_2': {'name': 'Bob'}}
高性能去重（适合大数据量）
def batch_dedup(data_list, key_func=None, chunk_size=1000):
"""
分批去重，适合大量数据
"""
if key_func is None:
key_func = lambda x: x
seen = set()
result = []
# 分批处理避免内存溢出
for i in range(0, len(data_list), chunk_size):
chunk = data_list[i:i + chunk_size]
for item in chunk:
key = key_func(item)
if key not in seen:
seen.add(key)
result.append(item)
return result
# 使用生成器版本
def dedup_generator(data_list, key_func=None):
"""
生成器版本，适合流式处理
"""
if key_func is None:
key_func = lambda x: x
seen = set()
for item in data_list:
key = key_func(item)
if key not in seen:
seen.add(key)
yield item
综合示例：缓存管理系统
class CacheDeduplicator:
def __init__(self, strategy='last'):
"""
strategy: 'first' - 保留首次出现
'last' - 保留最后出现
'merge' - 合并相同键的数据
"""
self.strategy = strategy
self.cache = {}
self.index = {} # 维护插入顺序
def dedup_list(self, data_list, key_field=None):
"""
对列表数据进行去重
"""
if not data_list:
return []
# 获取键函数
if key_field:
key_func = lambda x: x[key_field] if isinstance(x, dict) else x
else:
key_func = lambda x: x
if self.strategy == 'first':
return self._dedup_first(data_list, key_func)
elif self.strategy == 'last':
return self._dedup_last(data_list, key_func)
elif self.strategy == 'merge':
return self._dedup_merge(data_list, key_func)
else:
raise ValueError(f"Unknown strategy: {self.strategy}")
def _dedup_first(self, data_list, key_func):
"""保留首次出现"""
seen = {}
result = []
for item in data_list:
key = key_func(item)
if key not in seen:
seen[key] = True
result.append(item)
return result
def _dedup_last(self, data_list, key_func):
"""保留最后出现"""
seen = {}
for item in data_list:
key = key_func(item)
seen[key] = item
return list(seen.values())
def _dedup_merge(self, data_list, key_func):
"""合并相同键的数据"""
merged = {}
for item in data_list:
key = key_func(item)
if key not in merged:
merged[key] = item.copy() if isinstance(item, dict) else item
else:
if isinstance(item, dict) and isinstance(merged[key], dict):
merged[key].update(item)
# 非字典类型的合并规则自定义
return list(merged.values())
# 使用示例
deduplicator = CacheDeduplicator(strategy='last')
# 测试数据
data = [
{'id': 1, 'name': 'Alice', 'email': 'alice@example.com'},
{'id': 2, 'name': 'Bob', 'email': 'bob@example.com'},
{'id': 1, 'name': 'Alice Updated', 'email': 'alice.new@example.com'},
{'id': 3, 'name': 'Charlie'},
]
# 按id去重，保留最后出现的数据
result = deduplicator.dedup_list(data, key_field='id')
for item in result:
print(item)
使用建议
- 小数据量：使用字典或集合方法即可
- 大数据量：考虑分批处理或使用生成器
- 复杂对象：实现自定义的key_func
- 需要保持顺序：使用OrderedDict或首次出现策略
- 需要合并字段：使用merge策略
选择哪种方法取决于你的具体需求：数据量大小、是否需要保留顺序、是保留首次还是最后一次出现的数据等。
