如何用Python按列合并两个3.3G、3837万行的大型文件？

嘿，我来帮你搞定这个大文件合并的问题！先看看你这段代码里的几个明显问题，再给你优化出适合大文件的解决方案～

首先，你的代码里的3个关键错误 #

• 分块读取的写法错了：你直接用pd.read_csv()把整个a文件读进内存了，之后写a(chunksize=chunksize)完全不对——分块读取要在read_csv的时候就指定chunksize参数，这样返回的才是可以循环迭代的分块对象。
• 写入文件会覆盖结果：每次循环执行merged.to_csv("output.csv")会直接覆盖之前的内容，最后你只能得到最后一个分块的合并结果，必须用追加模式mode='a'，还要控制表头只写一次。
• 直接加载大文件b容易爆内存：3.3G的文件直接读进内存，要是你的内存不够大（比如8G内存），很容易直接卡死，这是处理大文件的大忌。

优化方案1：内存能放下b文件时（简单高效） #

如果你的电脑内存足够（比如16G以上），可以把b文件加载后设置Bin_ID为索引，这样合并的时候速度会快很多，同时分块读取a文件，逐块合并后追加写入：

import pandas as pd
import sys

# 读取b文件并设置Bin_ID为索引，提升merge效率
b = pd.read_csv(sys.argv[2], sep='\t', encoding="utf-8-sig", index_col='Bin_ID')

# 分块大小可以根据你的内存调整，100万行如果内存吃紧可以改小到50万
chunksize = 10**6
first_write = True  # 标记是否第一次写入，用来控制表头

# 正确的分块读取方式：在read_csv时指定chunksize
for chunk in pd.read_csv(sys.argv[1], sep='\t', encoding="utf-8-sig", chunksize=chunksize):
    # 合并两个数据块，因为b已经设了索引，merge会自动匹配，速度更快
    merged = chunk.merge(b, on='Bin_ID')
    # 写入文件：第一次写表头，之后追加时不写表头
    merged.to_csv("output.csv", index=False, sep='\t', mode='a', header=first_write)
    # 写完第一次后，把标记改为False
    first_write = False

优化方案2：内存放不下b文件时（超大型文件处理） #

如果你的内存连b文件都装不下，那就得用先排序再归并的思路（类似归并排序的原理），先借助命令行工具对两个文件按Bin_ID排序（比pandas排序快得多），再分块合并：

第一步：用命令行排序（Linux/macOS）

# 对a文件按第一列（Bin_ID）排序，分隔符是制表符，输出到sorted_a.tsv
sort -t $'\t' -k1,1 your_a_file.tsv > sorted_a.tsv
# 对b文件做同样操作
sort -t $'\t' -k1,1 your_b_file.tsv > sorted_b.tsv

第二步：pandas分块归并合并

import pandas as pd
import sys

chunksize = 10**6
# 打开两个已排序文件的分块读取器
a_reader = pd.read_csv("sorted_a.tsv", sep='\t', encoding="utf-8-sig", chunksize=chunksize)
b_reader = pd.read_csv("sorted_b.tsv", sep='\t', encoding="utf-8-sig", chunksize=chunksize)

# 初始化第一个块
a_chunk = next(a_reader)
b_chunk = next(b_reader)
first_write = True

while True:
    # 确定两个块中Bin_ID的交集范围
    current_min = max(a_chunk['Bin_ID'].min(), b_chunk['Bin_ID'].min())
    current_max = min(a_chunk['Bin_ID'].max(), b_chunk['Bin_ID'].max())
    
    if current_min > current_max:
        # 没有交集时，移动数据块较小的那个
        if a_chunk['Bin_ID'].max() < b_chunk['Bin_ID'].max():
            try:
                a_chunk = next(a_reader)
            except StopIteration:
                break
        else:
            try:
                b_chunk = next(b_reader)
            except StopIteration:
                break
        continue
    
    # 筛选出当前范围内的行进行合并
    a_filtered = a_chunk[(a_chunk['Bin_ID'] >= current_min) & (a_chunk['Bin_ID'] <= current_max)]
    b_filtered = b_chunk[(b_chunk['Bin_ID'] >= current_min) & (b_chunk['Bin_ID'] <= current_max)]
    merged = a_filtered.merge(b_filtered, on='Bin_ID')
    
    # 写入结果，控制表头
    merged.to_csv("output.csv", index=False, sep='\t', mode='a', header=first_write)
    first_write = False
    
    # 移除已经处理过的行，保留剩余行
    a_chunk = a_chunk[a_chunk['Bin_ID'] > current_max]
    b_chunk = b_chunk[b_chunk['Bin_ID'] > current_max]
    
    # 如果当前块为空，读取下一个块
    if a_chunk.empty:
        try:
            a_chunk = next(a_reader)
        except StopIteration:
            break
    if b_chunk.empty:
        try:
            b_chunk = next(b_reader)
        except StopIteration:
            break

这个方法的好处是不需要把整个文件放进内存，适合处理超大型数据集，就是多了一步排序的操作，但命令行的sort处理大文件效率非常高。

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