本文介绍一种高效、内存友好的方法，用于合并两个按时间排序的文本日志文件，在存在末尾-开头重叠时自动去重，同时严格保持原有时间顺序，避免全量读取或重复排序。
在处理分段采集的日志、传感器数据或时间序列记录（如 Jan-Mar.txt 和 Mar-Jun.txt）时，常需将两个已排序的文本文件合并为一个连续、无重复、仍保持严格时间顺序的文件。关键挑战在于：重叠只可能发生在第一个文件末尾与第二个文件开头之间，且重叠行内容完全一致；而文件体积很大，不能简单加载全部行后用 set 去重（会丢失顺序）或全量扫描去重（低效）。
上述答案中使用 OrderedDict.fromkeys(lines1 + lines2) 的方法虽简洁，但存在严重缺陷：
✅ 保留了首次出现的顺序（对 lines1 + lines2 拼接后去重）；
❌ 却无法保证时间顺序正确性——若 file2 开头几行早于 file1 结尾（即数据乱序），该方法仍会错误保留；
❌ 更严重的是，它将整个文件读入内存（list(…)），对“非常大的文件”不友好；
❌ 且未利用“两文件各自有序”这一关键前提，造成不必要的开销。
✅ 推荐方案：双指针流式合并（O(n+m) 时间，O(1) 额外内存）
我们采用类似归并排序的双指针策略，逐行读取、比较、写入，仅在重叠处跳过重复行：
def merge_sorted_files(file1_path, file2_path, output_path): """ 合并两个按时间升序排列的csv格式文本文件，自动跳过首尾重叠行。 假设每行以 ISO 格式时间戳开头（如 '2024-01-26 09:00:00,'），且重叠部分完全一致。 """ with open(file1_path, 'r') as f1, open(file2_path, 'r') as f2, open(output_path, 'w') as out: # 读取 file1 所有行（因需从末尾比对，且通常较小） lines1 = [line.rstrip('nr') for line in f1] # 逐行读取 file2，跳过与 lines1 尾部重复的行 skip_count = 0 for line2 in f2: line2 = line2.rstrip('nr') # 若 lines1 非空，且当前 line2 等于 lines1 最后一行，则跳过（重叠） if lines1 and line2 == lines1[-1]: skip_count += 1 continue # 否则，一旦发现不重叠，后续所有行均写入（因 file2 有序） break else: # file2 全部被跳过（极端情况：file2 是 file1 的后缀） line2 = None # 写入 file1 全部内容 out.writelines(f"{line}n" for line in lines1) # 写入 file2 中未被跳过的剩余行（包括刚读到的 line2） if line2 is not None: out.write(f"{line2}n") out.writelines(f"{line.rstrip('nr')}n" for line in f2) # 使用示例 merge_sorted_files('1.txt', '2.txt', 'merged.txt')✅ 进阶优化：真正流式处理（适用于超大文件）
若 file1 也极大（无法全量加载），可仅读取其最后一行用于比对：
def get_last_line(filepath): """安全获取文件最后一行（不加载全文）""" with open(filepath, 'rb') as f: f.seek(0, 2) # 移动到末尾 if f.tell() == 0: return "" f.seek(-1, 2) while f.tell() > 0 and f.read(1) != b'n': f.seek(-2, 1) return f.readline().decode().rstrip('nr') def stream_merge(file1_path, file2_path, output_path): last_line1 = get_last_line(file1_path) with open(file1_path, 'r') as f1, open(file2_path, 'r') as f2, open(output_path, 'w') as out: # 复制 file1 全部内容 for line in f1: out.write(line) # 跳过 file2 中与 last_line1 相同的开头行（最多一次重叠） for line in f2: stripped = line.rstrip('nr') if stripped == last_line1: continue # 跳过重叠行 out.write(line) break # 第一个非重叠行写入后，直接复制剩余全部 else: return # file2 已空或全重叠 # 复制剩余行 out.writelines(f2)⚠️ 注意事项
- 时间戳解析非必需：本方案依赖“重叠行内容完全相同”，无需解析日期——这正是题干给出的关键约束（“exactly the same data”），也是高效性的核心来源；
- 确保编码一致：打开文件时建议显式指定 encoding=’utf-8’；
- 原地覆盖风险：output_path 不应与 file1_path 或 file2_path 相同，避免数据损坏；
- 空行/空白处理：示例假设无空行；如有，可在读取时添加 if line.strip(): 过滤。
该方法时间复杂度为 O(n + m)，空间复杂度接近 O(1)（仅缓存一行或最后一行），完美契合“大文件、有序、局部重叠”的真实场景，是工业级日志合并的推荐实践。