本文目录导读：
我来介绍几种Python过滤冗余数据的常用方法：
基于哈希值去重
import hashlib
import json
class DataDeduplicator:
def __init__(self):
self.seen_hashes = set()
def get_data_hash(self, data):
"""生成数据哈希值"""
# 将数据转换为JSON字符串再计算哈希
if isinstance(data, (dict, list)):
data_str = json.dumps(data, sort_keys=True)
else:
data_str = str(data)
return hashlib.md5(data_str.encode()).hexdigest()
def is_duplicate(self, data):
"""检查数据是否重复"""
data_hash = self.get_data_hash(data)
if data_hash in self.seen_hashes:
return True
self.seen_hashes.add(data_hash)
return False
def filter_duplicates(self, data_list):
"""过滤重复数据"""
return [data for data in data_list if not self.is_duplicate(data)]
# 使用示例
dedup = DataDeduplicator()
data_list = [
{"id": 1, "name": "Alice"},
{"id": 2, "name": "Bob"},
{"id": 1, "name": "Alice"}, # 重复数据
]
filtered_data = dedup.filter_duplicates(data_list)
print(filtered_data)
基于时间戳过滤
from datetime import datetime, timedelta
import time
class TimeBasedFilter:
def __init__(self, time_window_seconds=3600):
self.time_window = timedelta(seconds=time_window_seconds)
self.last_sync_times = {}
def should_sync(self, data_id, data_timestamp):
"""
判断是否需要同步
Args:
data_id: 数据唯一标识
data_timestamp: 数据时间戳
"""
if data_id not in self.last_sync_times:
self.last_sync_times[data_id] = data_timestamp
return True
last_sync = self.last_sync_times[data_id]
time_diff = data_timestamp - last_sync
# 如果时间差大于时间窗，才需要同步
if timedelta(seconds=abs(time_diff.total_seconds())) > self.time_window:
self.last_sync_times[data_id] = data_timestamp
return True
return False
# 使用示例
filter = TimeBasedFilter(time_window_seconds=300) # 5分钟窗口
now = datetime.now()
data_items = [
("file1", now - timedelta(minutes=10)), # 需要同步
("file1", now - timedelta(minutes=2)), # 不需要同步
("file2", now - timedelta(minutes=30)), # 需要同步
]
for data_id, timestamp in data_items:
if filter.should_sync(data_id, timestamp):
print(f"需要同步: {data_id}")
else:
print(f"跳过同步: {data_id}")
基于版本号过滤
class VersionBasedFilter:
def __init__(self):
self.latest_versions = {}
def update_version(self, data_id, new_version):
"""更新数据版本"""
self.latest_versions[data_id] = new_version
def is_newer_version(self, data_id, version):
"""检查是否为更新版本"""
if data_id not in self.latest_versions:
return True
return version > self.latest_versions[data_id]
class DataItem:
def __init__(self, id, version, content):
self.id = id
self.version = version
self.content = content
# 使用示例
version_filter = VersionBasedFilter()
datasets = [
DataItem("doc1", 1, "版本1内容"),
DataItem("doc1", 2, "版本2内容"), # 新版本
DataItem("doc1", 1, "旧版本内容"), # 旧版本，应过滤
]
for dataset in datasets:
if version_filter.is_newer_version(dataset.id, dataset.version):
print(f"同步数据: {dataset.id} 版本 {dataset.version}")
version_filter.update_version(dataset.id, dataset.version)
else:
print(f"跳过旧版本: {dataset.id} 版本 {dataset.version}")
完整的数据同步过滤系统
import hashlib
import json
from datetime import datetime
from typing import Any, Dict, List, Optional
class SyncFilter:
"""同步过滤系统"""
def __init__(self):
self.seen_hashes = set() # 数据哈希
self.latest_versions = {} # 版本号
self.last_sync_times = {} # 同步时间
self.change_log = {} # 变更日志
def filter_by_hash(self, data: Any) -> bool:
"""基于哈希值过滤"""
data_hash = self._compute_hash(data)
if data_hash in self.seen_hashes:
return False
self.seen_hashes.add(data_hash)
return True
def filter_by_version(self, data_id: str, version: int) -> bool:
"""基于版本号过滤"""
current_version = self.latest_versions.get(data_id, -1)
if version <= current_version:
return False
self.latest_versions[data_id] = version
return True
def filter_by_time(self, data_id: str, timestamp: datetime,
window_minutes: int = 60) -> bool:
"""基于时间戳过滤"""
last_sync = self.last_sync_times.get(data_id)
if last_sync:
time_diff = (timestamp - last_sync).total_seconds() / 60
if time_diff < window_minutes:
return False
self.last_sync_times[data_id] = timestamp
return True
def filter_by_content(self, old_content: Any, new_content: Any) -> bool:
"""基于内容变化过滤"""
return old_content != new_content
def multi_filter(self, data: Dict) -> bool:
"""
多重过滤，任何一条过滤条件通过即认为需要同步
Returns:
True 表示需要同步，False 表示可以跳过
"""
data_id = data.get('id')
version = data.get('version', 0)
timestamp = data.get('timestamp', datetime.now())
content = data.get('content')
# 基于哈希值过滤（内容完全相同）
if not self.filter_by_hash(content):
return False
# 基于版本号过滤
if not self.filter_by_version(data_id, version):
return False
# 基于时间过滤（默认1小时内不重复同步）
if not self.filter_by_time(data_id, timestamp):
# 但如果版本有更新，仍然需要同步
if not self._has_version_update(data_id, version):
return False
return True
def _compute_hash(self, data: Any) -> str:
"""计算数据哈希"""
if isinstance(data, (dict, list)):
data_str = json.dumps(data, sort_keys=True)
else:
data_str = str(data)
return hashlib.md5(data_str.encode()).hexdigest()
def _has_version_update(self, data_id: str, version: int) -> bool:
"""检查版本是否有更新"""
current = self.latest_versions.get(data_id, -1)
return version > current
def log_sync(self, data_id: str, action: str, details: Optional[Dict] = None):
"""记录同步日志"""
if data_id not in self.change_log:
self.change_log[data_id] = []
self.change_log[data_id].append({
'action': action,
'timestamp': datetime.now(),
'details': details or {}
})
# 使用示例
def main():
sync_filter = SyncFilter()
# 模拟数据流
mock_data = [
{
'id': 'user_001',
'version': 1,
'timestamp': datetime(2024, 1, 1, 10, 0),
'content': {'name': 'Alice', 'email': 'alice@example.com'}
},
# 相同内容的重复数据
{
'id': 'user_001',
'version': 1,
'timestamp': datetime(2024, 1, 1, 10, 30),
'content': {'name': 'Alice', 'email': 'alice@example.com'}
},
# 新版本数据
{
'id': 'user_001',
'version': 2,
'timestamp': datetime(2024, 1, 1, 11, 0),
'content': {'name': 'Alice', 'email': 'alice_new@example.com'}
},
]
for i, data in enumerate(mock_data, 1):
print(f"\n数据 {i}:")
print(f" 内容: {data['content']}")
if sync_filter.multi_filter(data):
print(" ✅ 需要同步")
sync_filter.log_sync(data['id'], 'sync',
{'version': data['version']})
else:
print(" ❌ 跳过同步（冗余数据）")
if __name__ == "__main__":
main()
实用工具函数
def batch_filter_with_progress(data_list, filter_func, batch_size=100):
"""
批量过滤并显示进度
Args:
data_list: 数据列表
filter_func: 过滤函数
batch_size: 批处理大小
"""
total = len(data_list)
result = []
for i in range(0, total, batch_size):
batch = data_list[i:i + batch_size]
progress = min(i + batch_size, total)
print(f"处理进度: {progress}/{total}")
for data in batch:
if filter_func(data):
result.append(data)
return result
# 使用示例
def smart_filter(data):
"""智能过滤规则"""
# 过滤掉无效数据
if not data.get('id') or not data.get('content'):
return False
# 过滤掉空内容
if not data['content'].strip():
return False
# 过滤掉特定类型的数据
if data.get('type') == 'temporary':
return False
return True
data_list = [
{'id': '1', 'content': 'valid data'},
{'id': '2', 'content': ''}, # 空内容
{'type': 'temporary'}, # 临时数据
{'id': '3', 'content': 'another valid data'},
]
filtered = batch_filter_with_progress(data_list, smart_filter)
print(f"\n过滤结果: {filtered}")
选择哪种过滤方法取决于你的具体场景：
- 数据量大：使用哈希去重
- 有时间要求：使用时间戳过滤
- 版本管理：使用版本号过滤
- 综合需求：组合使用多种过滤方法
