Python 中实现多文件逐行合并与键值聚合的内存高效方案

聖光之護

发布时间：2026-02-13 21:58:01

800人浏览过

来源于php中文网

原创

Python 中实现多文件逐行合并与键值聚合的内存高效方案

本文介绍如何在不将整个文件加载到内存的前提下，对多个已排序的大文本文件进行逐行读取、k 路归并及相同键的值累加，适用于日志聚合、分布式计算结果合并等场景。

本文介绍如何在不将整个文件加载到内存的前提下，对多个已排序的大文本文件进行逐行读取、k 路归并及相同键的值累加，适用于日志聚合、分布式计算结果合并等场景。

在处理大规模键值数据（如 MapReduce 输出、分片日志统计）时，常遇到多个已按 key 排序的文本文件（每行格式为 key\tvalue），需将其合并为一个全局有序且同 key 值自动累加的结果。核心挑战在于：不能调用 file.readlines() 或 list(file) 全量加载，必须流式逐行读取；同时需支持多文件协同迭代，并对重复 key 进行归约（如求和）。

关键突破口在于理解 Python 文件对象的底层迭代机制：for line in file 实际隐式调用了 file.readline()，而显式调用 readline() 才能获得细粒度控制权——它每次只读取一行（含换行符），返回字符串或空字符串（表示 EOF），内存开销恒定，完全满足“大文件、低内存”需求。

以下是一个完整、健壮的 k 路合并实现：

import heapq
from typing import List, Tuple, Optional, TextIO

def merge_sorted_files(file_paths: List[str], output_path: str, sep: str = '\t') -> None:
    """
    合并多个已按 key 字典序排序的文本文件，对相同 key 的 value 执行数值累加。

    :param file_paths: 输入文件路径列表（每个文件每行格式为 "key<sep>value"）
    :param output_path: 输出文件路径
    :param sep: 键值分隔符，默认为 '\t'
    """
    # 打开所有输入文件，初始化文件句柄列表
    files = [open(path, 'r', encoding='utf-8') for path in file_paths]
    try:
        # 使用最小堆维护各文件当前行（key, value, file_index, line_content）
        heap = []
        for i, f in enumerate(files):
            line = f.readline()
            if line:  # 非空行才入堆
                key, val_str = line.rstrip('\n').split(sep, 1)
                try:
                    val = float(val_str)  # 支持整数/浮点数
                except ValueError:
                    raise ValueError(f"Invalid numeric value in {file_paths[i]}: {val_str}")
                heapq.heappush(heap, (key, val, i, line))

        # 归并主循环
        with open(output_path, 'w', encoding='utf-8') as out_f:
            while heap:
                curr_key, curr_val, idx, _ = heapq.heappop(heap)

                # 合并所有相同 key 的行（k 路归并中的“归约”阶段）
                merged_val = curr_val
                while heap and heap[0][0] == curr_key:
                    _, val, i, _ = heapq.heappop(heap)
                    merged_val += val
                    # 从对应文件读取下一行
                    next_line = files[i].readline()
                    if next_line:
                        k, v_str = next_line.rstrip('\n').split(sep, 1)
                        try:
                            v = float(v_str)
                        except ValueError:
                            raise ValueError(f"Invalid numeric value in {file_paths[i]}: {v_str}")
                        heapq.heappush(heap, (k, v, i, next_line))

                # 写入合并后结果
                out_f.write(f"{curr_key}{sep}{merged_val}\n")

                # 补充当前文件的下一行（若存在）
                next_line = files[idx].readline()
                if next_line:
                    k, v_str = next_line.rstrip('\n').split(sep, 1)
                    try:
                        v = float(v_str)
                    except ValueError:
                        raise ValueError(f"Invalid numeric value in {file_paths[idx]}: {v_str}")
                    heapq.heappush(heap, (k, v, idx, next_line))
    finally:
        # 确保所有文件正确关闭
        for f in files:
            f.close()

# 使用示例
if __name__ == "__main__":
    merge_sorted_files(
        file_paths=["part-00000", "part-00001", "part-00002"],
        output_path="merged_result.txt"
    )

✅ 关键设计说明：