Python实现含datetime.timedelta的数据集去重并按时间排序

我有一份数据，每行是datetime.timedelta类型的时间值和一个关联数值，需要高效去重并按时间值排序。当前用遍历判断的方法不仅没完成排序需求，处理2160万行、数GB级别的数据时效率极低，希望得到按时间有序的去重结果。

数据示例 #

(datetime.timedelta(seconds=34781, microseconds=474000), 0.004936)
(datetime.timedelta(seconds=33586, microseconds=443000), 0.003214)
...

尝试的低效代码 #

import datetime

listData = [(datetime.timedelta(seconds=34781, microseconds=474000), 0.004936),
            (datetime.timedelta(seconds=33586, microseconds=443000), 0.003214),
            (datetime.timedelta(seconds=34781, microseconds=474000), 0.004936),
            (datetime.timedelta(seconds=38306, microseconds=654000), 0.001765),
            (datetime.timedelta(seconds=38306, microseconds=654000), 0.001765),
            (datetime.timedelta(seconds=31245, microseconds=474000), 0.004938)]

finalTab = []
for i in listData:
   if i not in finalTab: finalTab.append(i)

print(finalTab)

期望输出 #

(datetime.timedelta(seconds=31245, microseconds=474000), 0.004938)
(datetime.timedelta(seconds=33586, microseconds=443000), 0.003214)
...

解决方案 #

1. 内存足够时：集合去重+排序 #

集合去重的时间复杂度为O(n)，远高于遍历判断的O(n²)，之后直接按timedelta值排序即可：

import datetime

listData = [(datetime.timedelta(seconds=34781, microseconds=474000), 0.004936),
            (datetime.timedelta(seconds=33586, microseconds=443000), 0.003214),
            (datetime.timedelta(seconds=34781, microseconds=474000), 0.004936),
            (datetime.timedelta(seconds=38306, microseconds=654000), 0.001765),
            (datetime.timedelta(seconds=38306, microseconds=654000), 0.001765),
            (datetime.timedelta(seconds=31245, microseconds=474000), 0.004938)]

# 集合去重后按timedelta排序
unique_sorted = sorted(set(listData), key=lambda x: x[0])

for item in unique_sorted:
    print(item)

• 原理：元组是可哈希类型，set()可自动去重；sorted()通过key指定按元组第一个元素（timedelta）排序，timedelta本身支持比较运算，无需额外转换。
• 优势：处理千万级数据时，效率比遍历判断提升几个数量级。

2. 内存不足时：分批处理+外部排序 #

若数据过大无法一次性载入内存，可通过以下步骤处理：

1. 分批读取：每次读入部分数据，去重后按timedelta排序，写入临时文件。
2. 多路归并：将多个有序临时文件做归并排序，得到全局有序的去重结果。

简化版代码框架：

import datetime
import os
import heapq

# 分批读取并处理大文件（需替换实际数据读取逻辑）
batch_size = 1000000
temp_files = []
batch_num = 0

with open("large_data.txt", "r") as f:
    batch = []
    for line in f:
        # 解析行数据（建议替换eval为手动解析，避免安全风险）
        delta_str, value_str = eval(line.strip())
        batch.append((delta_str, value_str))
        
        if len(batch) >= batch_size:
            # 去重+排序当前批次
            unique_batch = sorted(set(batch), key=lambda x: x[0])
            # 写入临时文件
            temp_file = f"temp_{batch_num}.txt"
            with open(temp_file, "w") as tf:
                tf.write("\n".join(str(item) for item in unique_batch))
            temp_files.append(temp_file)
            batch_num += 1
            batch = []
    # 处理剩余数据
    if batch:
        unique_batch = sorted(set(batch), key=lambda x: x[0])
        temp_file = f"temp_{batch_num}.txt"
        with open(temp_file, "w") as tf:
            tf.write("\n".join(str(item) for item in unique_batch))
        temp_files.append(temp_file)

# 多路归并临时文件
def merge_sorted_files(file_paths, output_path):
    file_handles = [open(path, "r") for path in file_paths]
    heap = []
    last_delta = None
    
    # 初始化堆：每个文件取第一行
    for idx, fh in enumerate(file_handles):
        line = fh.readline().strip()
        if line:
            item = eval(line)
            heapq.heappush(heap, (item[0], item, idx))
    
    with open(output_path, "w") as out_f:
        while heap:
            delta, item, idx = heapq.heappop(heap)
            # 全局去重
            if delta != last_delta:
                out_f.write(f"{item}\n")
                last_delta = delta
            # 读取当前文件下一行
            fh = file_handles[idx]
            line = fh.readline().strip()
            if line:
                next_item = eval(line)
                heapq.heappush(heap, (next_item[0], next_item, idx))
            else:
                fh.close()
    
    # 清理临时文件
    for path in file_paths:
        os.remove(path)

merge_sorted_files(temp_files, "final_result.txt")

• 注意：实际使用时需替换eval为更安全的字符串解析逻辑（比如拆分字符串后手动构造datetime.timedelta），避免代码注入风险并提升解析效率。

内容的提问来源于stack exchange，提问作者Hug