如何用Pandas或Python逐行合并大TXT文件优化内存占用

核心优化逻辑：完全规避全量加载大体积A.txt的内存开销，仅将体量更小的B.txt全量缓存为可快速查询的索引结构，逐行流式读取A.txt完成匹配拼接，全程内存占用仅与B.txt体量、单行数据大小挂钩。

实现步骤 #

• 读取体量较小的B.txt，将所有contig及对应的关联字段存为Python字典，实现O(1)时间复杂度的匹配查询。
• 对齐原pandas合并逻辑的输出列顺序，提前写入结果文件的表头。
• 逐行读取A.txt，每读取一行就检查该行的contig是否在预加载的字典中，匹配成功才将拼接后的行写入结果文件，未匹配的行直接跳过，不占用内存缓存。
• 全程无A.txt全量数据的内存驻留，可适配单文件几十GB级别的A.txt处理需求。

可直接运行的实现代码 #

import csv

# 加载体量较小的B.txt，构建contig匹配索引
coverage_index = {}
with open("B.txt", "r", encoding="utf-8", newline="") as file_b:
    reader_b = csv.DictReader(file_b, delimiter="\t")
    # 提取B表中除公共字段contig外的其余列
    extra_b_cols = [col for col in reader_b.fieldnames if col != "contig"]
    for row in reader_b:
        coverage_index[row["contig"]] = [row[col] for col in extra_b_cols]

# 流式读取A.txt并写出匹配结果
with open("A.txt", "r", encoding="utf-8", newline="") as file_a, \
     open("merged_file.txt", "w", encoding="utf-8", newline="") as file_out:
    reader_a = csv.DictReader(file_a, delimiter="\t")
    # 输出列顺序与原pandas inner merge结果完全一致
    output_cols = reader_a.fieldnames + extra_b_cols
    writer_out = csv.DictWriter(file_out, fieldnames=output_cols, delimiter="\t")
    writer_out.writeheader()

    # 逐行匹配处理
    for row_a in reader_a:
        contig_val = row_a["contig"]
        if contig_val in coverage_index:
            # 拼接当前A行的所有字段
            merged_row = {col: row_a[col] for col in reader_a.fieldnames}
            # 补充匹配到的B表字段
            for col_pos, col_name in enumerate(extra_b_cols):
                merged_row[col_name] = coverage_index[contig_val][col_pos]
            writer_out.writerow(merged_row)

实现兼容性说明 #

• 合并规则与原pd.merge(..., on='contig', how='inner')完全一致，输出结果与原实现无差异，仅保留A、B两表contig字段匹配的条目。
• 支持A表中同一contig对应多行的多对一匹配场景，与pandas的匹配逻辑完全对齐。
• 内置csv模块原生支持制表符分隔文件的解析，可自动处理字段内的特殊字符、转义场景，不会出现分割错位问题。
• 若B表的contig规模达到千万级，仅需不到1GB内存即可完成索引构建，远低于全量加载大A.txt的内存占用。

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