多差异行列文件合并：大内存数据高效处理方案求助

针对2200万行、1000列的超大数据量，以下是R、Python、Bash三种环境下的低内存高效解决方案：

一、Python 解决方案（推荐超大数据量场景） #

方法1：Dask DataFrame（分布式计算，完全规避内存瓶颈） #

Dask将数据拆分成小块并行处理，无需一次性加载全量数据到内存，适合超大规模数据集：

import dask.dataframe as dd

# 读取文件，指定SNP为索引，统一用字符串类型减少内存占用
df1 = dd.read_csv('file1.txt', sep='\t', index_col='SNP', dtype=str)
df2 = dd.read_csv('file2.txt', sep='\t', index_col='SNP', dtype=str)
df3 = dd.read_csv('file3.txt', sep='\t', index_col='SNP', dtype=str)

# 全外连接合并三个文件
merged = df1.join(df2, how='outer').join(df3, how='outer')

# 合并重复的公共列（如CHROM、POS），取第一个非缺失值
common_cols = ['CHROM', 'POS', 'ref', 'alt', 'GT', 'info']
for col in common_cols:
    # 收集所有包含该列名的字段
    related_cols = [c for c in merged.columns if col in c]
    # 按行填充缺失值，保留第一个有效内容
    merged[col] = merged[related_cols].bfill(axis=1).iloc[:, 0]
    # 删除重复列
    merged = merged.drop(related_cols[1:], axis=1)

# 填充所有缺失值为./.
merged = merged.fillna('./.')

# 导出为单个文本文件
merged.to_csv('merged_result.txt', sep='\t', single_file=True, index=True)

方法2：Pandas 分块处理（无需额外框架） #

通过分块读取文件，先收集所有唯一SNP，再逐步填充数据：

import pandas as pd

# 第一步：收集所有唯一SNP
all_snps = set()
for file in ['file1.txt', 'file2.txt', 'file3.txt']:
    for chunk in pd.read_csv(file, sep='\t', usecols=['SNP'], chunksize=100000):
        all_snps.update(chunk['SNP'].tolist())
all_snps = list(all_snps)

# 获取所有列名（公共列+所有样本列）
all_cols = []
common_cols = []
for file in ['file1.txt', 'file2.txt', 'file3.txt']:
    header = pd.read_csv(file, sep='\t', nrows=0).columns.tolist()
    if not common_cols:
        common_cols = [col for col in header if 'sample' not in col]
    all_cols.extend([col for col in header if 'sample' in col])
all_cols = common_cols + list(dict.fromkeys(all_cols))  # 去重样本列

# 初始化结果表，默认填充./.
result = pd.DataFrame(index=all_snps, columns=all_cols).fillna('./.')

# 分块读取每个文件，填充对应数据
for file in ['file1.txt', 'file2.txt', 'file3.txt']:
    for chunk in pd.read_csv(file, sep='\t', chunksize=100000, dtype=str):
        chunk.set_index('SNP', inplace=True)
        # 仅更新存在的行和列
        result.loc[chunk.index, chunk.columns] = chunk.values

# 导出结果
result.to_csv('merged_result.txt', sep='\t')

二、R 解决方案（优化内存占用） #

改用data.table包，其内存效率远高于dplyr，适合大文件处理：

library(data.table)

# 读取文件时指定SNP为键，统一字符类型减少内存
dt1 <- fread("file1.txt", key = "SNP", colClasses = rep("character", ncol(fread("file1.txt", nrows=0))))
dt2 <- fread("file2.txt", key = "SNP", colClasses = rep("character", ncol(fread("file2.txt", nrows=0))))
dt3 <- fread("file3.txt", key = "SNP", colClasses = rep("character", ncol(fread("file3.txt", nrows=0))))

# 全外连接合并，基于公共列匹配
merged <- merge(dt1, dt2, by = c("SNP", "CHROM", "POS", "ref", "alt", "GT", "info"), all = TRUE)
merged <- merge(merged, dt3, by = c("SNP", "CHROM", "POS", "ref", "alt", "GT", "info"), all = TRUE)

# 填充所有缺失值为./.
merged[is.na(merged)] <- "./."

# 用fwrite快速导出，比write.table效率高
fwrite(merged, "merged_result.txt", sep = "\t")

进阶分块处理（极端内存不足场景） #

library(data.table)

# 收集所有唯一SNP
all_snps <- c()
for (file in c("file1.txt", "file2.txt", "file3.txt")) {
  chunk_reader <- fread(file, select = "SNP", chunksize = 1e5)
  while (!chunk_reader$isDone()) {
    chunk <- chunk_reader$read()
    all_snps <- unique(c(all_snps, chunk$SNP))
  }
}

# 获取所有列名
all_cols <- c()
common_cols <- c()
for (file in c("file1.txt", "file2.txt", "file3.txt")) {
  header <- colnames(fread(file, nrows = 0))
  if (length(common_cols) == 0) {
    common_cols <- header[!grepl("sample", header)]
  }
  all_cols <- unique(c(all_cols, header))
}

# 初始化结果表，默认填充./.
result <- data.table(SNP = all_snps)
result[, (setdiff(all_cols, "SNP")) := "./."]
setkey(result, "SNP")

# 分块读取文件并填充数据
for (file in c("file1.txt", "file2.txt", "file3.txt")) {
  chunk_reader <- fread(file, chunksize = 1e5, colClasses = "character")
  while (!chunk_reader$isDone()) {
    chunk <- chunk_reader$read()
    setkey(chunk, "SNP")
    # 更新匹配行的对应列
    result[chunk, (colnames(chunk)) := mget(colnames(chunk))]
  }
}

# 导出结果
fwrite(result, "merged_result.txt", sep = "\t")

三、Bash 解决方案（Linux命令行环境） #

利用sort和join命令高效处理，无需编程语言环境：

# 1. 按SNP列排序所有文件（join要求输入有序）
sort -k1,1 file1.txt > sorted_file1.txt
sort -k1,1 file2.txt > sorted_file2.txt
sort -k1,1 file3.txt > sorted_file3.txt

# 2. 合并前两个文件，填充缺失值
join -a1 -a2 -t$'\t' -o auto sorted_file1.txt sorted_file2.txt | awk '
BEGIN {OFS="\t"}
NR==1 {print $0; next}
{
    # 填充所有空值为./.
    for (i=1; i<=NF; i++) {
        if ($i == "") $i = "./."
    }
    print $0
}' > merged_temp.txt

# 3. 合并第三个文件到临时结果
sort -k1,1 merged_temp.txt > sorted_merged_temp.txt
join -a1 -a2 -t$'\t' -o auto sorted_merged_temp.txt sorted_file3.txt | awk '
BEGIN {OFS="\t"}
NR==1 {print $0; next}
{
    for (i=1; i<=NF; i++) {
        if ($i == "") $i = "./."
    }
    print $0
}' > merged_result.txt

# 清理临时文件
rm sorted_file1.txt sorted_file2.txt sorted_file3.txt merged_temp.txt sorted_merged_temp.txt

注：Bash方法需确保公共列（如CHROM、POS）在各文件中值一致，若存在冲突需额外处理。

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