如何高效合并两个有序文本文件并自动去重（基于行内容，保留时序）_技术教程

本文介绍一种高效、内存友好的方法，用于合并两个按时间排序的文本日志文件，在存在末尾-开头重叠时自动去重，同时严格保持原有时间顺序，避免全量读取或重复排序。

在处理分段采集的日志、传感器数据或时间序列记录（如 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 格式时间戳开头（如 '2025-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('\n\r') for line in f1]

        # 逐行读取 file2，跳过与 lines1 尾部重复的行
        skip_count = 0
        for line2 in f2:
            line2 = line2.rstrip('\n\r')
            # 若 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('\n\r')}\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('\n\r')

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('\n\r')
            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)（仅缓存一行或最后一行），完美契合“大文件、有序、局部重叠”的真实场景，是工业级日志合并的推荐实践。