Python脚本如何去重整合缓存重复数据

本文目录导读：

1. 基于字典的去重（保留最后出现的数据）
2. 基于集合的去重（保留首次出现的数据）
3. 使用OrderedDict保持插入顺序
4. 缓存数据整合去重（复杂对象）
5. 高性能去重（适合大数据量）
6. 综合示例：缓存管理系统
7. 使用建议

我来介绍几种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)

使用建议

1. 小数据量：使用字典或集合方法即可
2. 大数据量：考虑分批处理或使用生成器
3. 复杂对象：实现自定义的key_func
4. 需要保持顺序：使用OrderedDict或首次出现策略
5. 需要合并字段：使用merge策略

选择哪种方法取决于你的具体需求：数据量大小、是否需要保留顺序、是保留首次还是最后一次出现的数据等。