如何在Snakemake中实现文件夹的批量拆分（按固定文件数）？

你可以用checkpoint结合自定义文件拆分函数来实现这个需求——checkpoint的核心就是处理动态生成的输出，刚好适配这种根据输入文件数量自动调整子文件夹数量的场景。以下是可复用的实现方案：

1. 核心思路 #

• 用checkpoint任务扫描目标文件夹的所有文件，按指定数量（比如3000个/文件夹）自动计算需要拆分的子文件夹数量
• 批量将文件复制/移动到对应的子文件夹中
• 后续分析任务动态获取这些子文件夹，并行处理每个子文件夹内的文件

2. 具体代码实现 #

第一步：定义文件拆分函数 #

先写一个通用的拆分函数，负责计算子文件夹数量、创建文件夹并分配文件：

import os
import glob
import shutil

def split_files_into_dirs(input_dir, files_per_dir=3000):
    # 获取目标文件夹下的所有文件（可根据需求修改glob规则过滤后缀）
    all_files = sorted(glob.glob(os.path.join(input_dir, "*")))
    if not all_files:
        raise ValueError(f"Input directory {input_dir} contains no files")
    
    # 向上取整计算需要的子文件夹数量
    num_subdirs = (len(all_files) + files_per_dir - 1) // files_per_dir
    split_root = f"{input_dir}_split"
    os.makedirs(split_root, exist_ok=True)
    
    split_dirs = []
    for idx in range(num_subdirs):
        # 生成子文件夹路径（按序号命名，保证顺序）
        subdir_path = os.path.join(split_root, f"batch_{idx:03d}")
        os.makedirs(subdir_path, exist_ok=True)
        
        # 分配当前批次的文件
        start_idx = idx * files_per_dir
        end_idx = start_idx + files_per_dir
        batch_files = all_files[start_idx:end_idx]
        
        # 复制文件到子文件夹（如需移动文件，替换为shutil.move）
        for f in batch_files:
            shutil.copy(f, subdir_path)
        
        split_dirs.append(subdir_path)
    
    return split_dirs

第二步：定义Checkpoint任务 #

用checkpoint执行拆分逻辑，确保Snakemake能追踪动态生成的子文件夹：

checkpoint split_input_folder:
    input:
        input_dir="path/to/your/source_folder"  # 替换为你的目标文件夹路径
    output:
        directory("{input.input_dir}_split")  # 标记拆分后的根文件夹为输出
    run:
        # 执行拆分函数
        split_files_into_dirs(input.input_dir, files_per_dir=3000)
        # 创建标记文件，让Snakemake确认任务完成
        with open(os.path.join(output[0], ".split_complete"), "w") as f:
            f.write(f"Split {len(glob.glob(os.path.join(input.input_dir, '*')))} files into {len(os.listdir(output[0]))} batches\n")

第三步：动态获取拆分后的子文件夹并处理 #

定义一个辅助函数获取所有子文件夹，后续分析任务基于这些子文件夹并行执行：

def get_split_batches(wildcards):
    # 从checkpoint的输出中获取拆分后的子文件夹列表
    split_root = checkpoints.split_input_folder.get_output(wildcards)[0]
    return sorted(glob.glob(os.path.join(split_root, "batch_*")))

# 示例：处理每个子文件夹的任务
rule process_batch:
    input:
        batch_dir=lambda wildcards: get_split_batches(wildcards)[wildcards.batch_idx]
    output:
        "{batch_dir}/processing_done.txt"
    shell:
        """
        # 替换为你的实际分析命令，比如调用脚本处理当前批次的所有文件
        echo "Processing batch {input.batch_dir} at $(date)" > {output}
        # your_analysis_tool --input {input.batch_dir} --output {input.batch_dir}/results
        """

# 总任务：触发拆分+所有批次处理
rule all:
    input:
        expand("{batch_dir}/processing_done.txt", batch_dir=get_split_batches)

3. 关键细节说明 #

• 文件操作选择：用shutil.copy保留原文件，如需清理原文件夹，换成shutil.move即可
• 文件过滤：修改glob.glob的规则（比如os.path.join(input_dir, "*.fastq")）可只处理特定类型文件
• 自动适配数量：num_subdirs = (len(all_files) + files_per_dir -1) // files_per_dir 这个公式会自动向上取整，比如45000个文件会生成15个批次，45001个则生成16个
• 并行处理：Snakemake会自动并行处理所有process_batch任务，充分利用计算资源

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