本文介绍如何在内存受限场景下，高效合并多个已排序的大文本文件，每行含一个键值对，支持按键去重并累加值，核心在于使用 readline() 逐行控制读取。
本文介绍如何在内存受限场景下，高效合并多个已排序的大文本文件，每行含一个键值对，支持按键去重并累加值，核心在于使用 readline() 逐行控制读取。
在处理超大日志、分片统计或分布式计算输出时，常遇到多个已按 key 排序的纯文本文件（如 part-00001.txt, part-00002.txt），每行格式为 key\tvalue（如 user_123\t42）。目标是将所有文件流式合并：
✅ 保持全局 key 有序（因输入已排序）
✅ 相同 key 的 value 累加（而非覆盖）
✅ 不一次性加载整文件到内存（规避 readlines() 或 file.read()）
✅ 支持任意数量输入文件
关键突破口在于：file.readline() 是可控、低开销的逐行读取原语——它只读取当前行（含换行符），返回字符串或空字符串（EOF），完全满足“手动推进读取指针”的需求。
✅ 正确做法：k-way merge + readline 驱动
以下是一个健壮、可扩展的实现：
import heapq
from typing import Iterator, Tuple, TextIO, Optional
def merge_sorted_files(file_paths: list[str], sep: str = '\t') -> Iterator[Tuple[str, int]]:
"""
流式合并多个已排序的键值文件，相同 key 的 value 自动累加。
Args:
file_paths: 文件路径列表（按需排序）
sep: 键值分隔符，默认 '\t'
Yields:
(key, total_value) 元组，按 key 升序
"""
# 打开所有文件，初始化每文件的首行
files = [open(path, 'r') for path in file_paths]
# 堆元素：(key, value, file_index, file_handle)
heap = []
for i, f in enumerate(files):
line = f.readline().strip()
if line: # 跳过空行
try:
key, val_str = line.split(sep, 1)
value = int(val_str.strip())
heapq.heappush(heap, (key, value, i, f))
except (ValueError, IndexError):
raise ValueError(f"Invalid line format in {file_paths[i]}: {line}")
# k-way merge 主循环
while heap:
key, value, idx, f = heapq.heappop(heap)
# 合并所有相同 key 的项（因文件已排序，相同 key 必连续出现在堆顶）
while heap and heap[0][0] == key:
_, next_val, next_idx, next_f = heapq.heappop(heap)
value += next_val
# 推入 next_f 的下一行（若存在）
next_line = next_f.readline().strip()
if next_line:
try:
nk, nv_str = next_line.split(sep, 1)
nv = int(nv_str.strip())
heapq.heappush(heap, (nk, nv, next_idx, next_f))
except (ValueError, IndexError):
raise ValueError(f"Invalid line in {file_paths[next_idx]}: {next_line}")
yield (key, value)
# 推入当前文件的下一行
next_line = f.readline().strip()
if next_line:
try:
nk, nv_str = next_line.split(sep, 1)
nv = int(nv_str.strip())
heapq.heappush(heap, (nk, nv, idx, f))
except (ValueError, IndexError):
raise ValueError(f"Invalid line in {file_paths[idx]}: {next_line}")
# 关闭所有文件
for f in files:
f.close()
# 使用示例
if __name__ == "__main__":
# 假设有 3 个已排序文件：
# a.txt: apple\t10\n banana\t5\n
# b.txt: apple\t3\n cherry\t7\n
# c.txt: banana\t2\n cherry\t1\n
for key, total in merge_sorted_files(['a.txt', 'b.txt', 'c.txt']):
print(f"{key}\t{total}")
# 输出：
# apple 13
# banana 7
# cherry 8⚠️ 注意事项与最佳实践
- 必须保证输入文件严格按 key 字典序（或数值序）升序排列，否则合并结果错误；
- readline() 返回含 \n 的字符串，务必用 .strip() 清理，避免 key 包含空白符；
- 错误处理不可省略：空行、格式异常、非数字 value 应显式捕获并报错，便于调试；
- 若 value 类型为浮点数或自定义结构，需相应调整解析逻辑（如 float() 或 json.loads()）；
- 对于超大规模场景（>1000 文件），可考虑批量打开/关闭或使用上下文管理器（with）增强安全性；
- 内存占用 ≈ O(k)，其中 k 为文件数（堆中最多存 k 个元素），与文件总行数无关，真正实现“内存友好”。
✅ 总结
file.readline() 是实现外部归并的核心基石——它赋予你精确的逐行控制权，避免了 for line in file 的隐式迭代绑定。结合最小堆（heapq）实现 k-way merge，并在弹出时主动聚合相同 key，即可在恒定内存下完成高效、正确、可扩展的多文件合并。这不是“技巧”，而是处理大数据流水线的标准范式。
