NCBI GEO多源基因表达数据集自动化合并方案咨询

阿华AIGC实验室

2026-5-13

我之前处理过类似的大规模GEO数据集整合任务，给你一套可复用的自动化方案——核心是先解决基因ID标准化这个最棘手的问题，再逐步统一注释、批量处理合并：

核心解决方案：三步实现自动化整合

1. 统一基因ID：从混乱命名映射到权威标准

不同数据集的基因命名（基因名、locus tag、别名等）是合并的最大障碍，必须先把所有基因名映射到一个权威标准ID（比如NCBI Gene ID、Ensembl ID或HGNC符号，根据你的物种选择）。

具体实现（Python为例，R可替换为biomaRt）

用mygene库做批量ID映射，它支持多种基因标识符的交叉查询：

import mygene
import pandas as pd
import os

# 初始化mygene客户端
mg = mygene.MyGeneInfo()

# 第一步：收集所有文件中的基因名并去重
def collect_all_gene_names(file_dir):
    gene_names = set()
    for filename in os.listdir(file_dir):
        if filename.endswith('.txt'):  # 替换为你的文件格式
            df = pd.read_csv(os.path.join(file_dir, filename), sep='\t')
            # 假设基因名在第一列，或者根据列名定位，比如df['Gene']
            gene_names.update(df.iloc[:, 0].tolist())
    return list(gene_names)

all_genes = collect_all_gene_names('./your_dataset_dir')

# 第二步：批量映射到标准ID（以大肠杆菌为例，物种参数改为escherichia_coli；人类用human）
mapping_results = mg.querymany(
    all_genes,
    scopes=['symbol', 'alias', 'locustag', 'refseq'],  # 覆盖所有可能的命名类型
    fields=['entrezgene', 'symbol', 'locus_tag'],
    species='escherichia_coli',
    returnall=True,
    verbose=True
)

# 整理成映射字典，处理无匹配的情况
gene_mapping = {}
for item in mapping_results['out']:
    query_name = item['query']
    # 优先取NCBI Gene ID，没有的话保留原名称
    standard_id = item.get('entrezgene', query_name)
    gene_mapping[query_name] = standard_id

# 处理无匹配的基因（可选：存入单独文件记录）
unmapped = [gene for gene in all_genes if gene not in gene_mapping]
with open('./unmapped_genes.txt', 'w') as f:
    f.write('\n'.join(unmapped))

2. 统一注释列：标准化字段命名与结构

不同数据集的注释列命名混乱（比如Gene Name/gene_symbol/Symbol），需要先提取所有注释字段，统一命名规则：

实现步骤：

遍历所有文件，收集所有注释列名，建立同义映射（比如把Locus Tag/locus_tag都映射为locus_tag）
对每个文件，重命名注释列，填充缺失字段，分离注释列与表达列
给表达列加上数据集前缀，避免合并时列名冲突

代码片段：

# 定义注释列的映射规则，根据你收集的列名扩展
annotation_mapping = {
    'Gene Name': 'symbol',
    'Locus Tag': 'locus_tag',
    'Gene Description': 'description',
    'Chr': 'chromosome'
}

def process_single_file(file_path):
    df = pd.read_csv(file_path, sep='\t')
    # 重命名注释列
    df = df.rename(columns=annotation_mapping)
    # 标准化基因ID
    df['standard_gene_id'] = df.iloc[:, 0].map(gene_mapping).fillna(df.iloc[:, 0])
    # 分离注释列和表达列
    core_annotations = ['standard_gene_id', 'symbol', 'locus_tag', 'description', 'chromosome']
    # 过滤出存在的注释列，避免KeyError
    existing_annotations = [col for col in core_annotations if col in df.columns]
    expr_cols = [col for col in df.columns if col not in existing_annotations]
    # 给表达列加数据集前缀（用文件名区分）
    dataset_name = os.path.basename(file_path).split('.')[0]
    expr_col_rename = {col: f"{dataset_name}_{col}" for col in expr_cols}
    df = df.rename(columns=expr_col_rename)
    # 返回标准化后的注释+表达列
    return df[existing_annotations + list(expr_col_rename.values())]

3. 批量处理与合并：高效整合10000个文件

直接循环合并10000个文件会导致内存溢出，建议分阶段合并：

优化方案：分块合并+并行处理

from multiprocessing import Pool
import glob

# 第一步：批量处理所有文件，生成中间文件
file_list = glob.glob('./your_dataset_dir/*.txt')

def save_processed_file(file_path):
    processed_df = process_single_file(file_path)
    output_name = f"./processed/{os.path.basename(file_path)}"
    processed_df.to_csv(output_name, sep='\t', index=False)

# 用多进程并行处理，提升速度
with Pool(processes=8) as pool:  # 根据你的CPU核心数调整
    pool.map(save_processed_file, file_list)

# 第二步：分块合并中间文件（比如每100个文件合并为一个块）
processed_files = glob.glob('./processed/*.txt')
chunk_size = 100
chunks = [processed_files[i:i+chunk_size] for i in range(0, len(processed_files), chunk_size)]

def merge_chunk(chunk_files):
    chunk_df = pd.DataFrame()
    for file in chunk_files:
        df = pd.read_csv(file, sep='\t')
        if chunk_df.empty:
            chunk_df = df
        else:
            # 按standard_gene_id合并，保留所有字段
            chunk_df = pd.merge(chunk_df, df, on='standard_gene_id', how='outer', suffixes=('', '_dup'))
            # 处理重复的注释列，取第一个非空值
            for col in ['symbol', 'locus_tag', 'description', 'chromosome']:
                if f"{col}_dup" in chunk_df.columns:
                    chunk_df[col] = chunk_df[col].fillna(chunk_df[f"{col}_dup"])
                    chunk_df = chunk_df.drop(columns=[f"{col}_dup"])
    return chunk_df

# 合并所有块
chunk_dfs = []
for chunk in chunks:
    chunk_dfs.append(merge_chunk(chunk))

final_df = pd.DataFrame()
for chunk_df in chunk_dfs:
    if final_df.empty:
        final_df = chunk_df
    else:
        final_df = pd.merge(final_df, chunk_df, on='standard_gene_id', how='outer', suffixes=('', '_dup'))
        # 再次处理重复注释列
        for col in ['symbol', 'locus_tag', 'description', 'chromosome']:
            if f"{col}_dup" in final_df.columns:
                final_df[col] = final_df[col].fillna(final_df[f"{col}_dup"])
                final_df = final_df.drop(columns=[f"{col}_dup"])

# 保存最终整合文件
final_df.to_csv('./integrated_gene_expression.txt', sep='\t', index=False)