如何在Nextflow中调用Bash脚本并管理多输入

尝试将Bash脚本集成到Nextflow管道中，该脚本读取两个目录下的BED文件并执行交集操作。终端直接运行脚本可正常输出交集结果，但在Nextflow中运行时仅生成预期的空motifs_loc子目录，无交集文件输出。推测需将BED文件纳入输入通道，但不清楚如何处理多层目录结构，寻求实现指导。

运行前目录结构 #

results
├── extract_gimme
│   ├── gimme_scan_Ss1
│   ├── gimme_scan_Ss2
│   ├── gimme_scan_Ss3
│   ├── gimme_scan_Ss4
│   └── gimme_scan_Ss5
├── gimme_scan
├── motifs_nr_coord

原Bash脚本 #

# get list of extract_gimme/ & write to txt
ls $1 > dir_list.txt

echo "intersect starts here!"
echo "------------------------"
echo " "

#now run loop for each dir inside dir_list.txt
while read DIR;
do

# declare the two directories for intersect
bed_dir="$1$DIR"
ls $bed_dir > bed_files.txt
coord_dir="results/motifs_nr_coord"

#create directory to save outputs for each DIR
DIR_A="results/motifs_loc"
DIR_B=$(echo $DIR| cut -d'_' -f 3) #this will  get Ss[12345]
motif_dir="$DIR_A"/"$DIR_B"
mkdir -p $motif_dir # to output results

# process each sub dir using loop & run intersect
  while read BED;
  do
   bedtools intersect -a $bed_dir"/"$BED -b $coord_dir"/"$BED -wa | sort | uniq > $motif_dir"/"$BED"_intersected.bed"
  done < bed_files.txt

rm bed_files.txt
echo "sample processing for $DIR_B done!!"
echo " "
echo "**********************"

done  < dir_list.txt
rm dir_list.txt

原Nextflow管道代码 #

#!/usr/bin/env nextflow
nextflow.enable.dsl=2

params.gimme_scan_dir = "results/extract_gimme"

process INTERSECT {
    publishDir "results", mode: 'copy', overwrite: false

    input:
    path scan_dir
    
    output:
    path '*'

    script:
    """
    $baseDir/intersect_scan_coord.sh $scan_dir
    """
}

workflow {
    scan_dir = Channel.fromPath(params.gimme_scan_dir, type: 'dir')
    INTERSECT(scan_dir)
}

问题分析 #

1. 路径依赖错误：原脚本硬编码了results/motifs_nr_coord相对路径，但Nextflow进程在隔离的工作目录中运行，该目录不会自动被传入工作环境，导致脚本找不到配对的BED文件。
2. 输入不完整：仅传入了extract_gimme目录，未将motifs_nr_coord目录作为输入传入进程，脚本无法获取交集所需的B文件。
3. 输出捕获失效：原输出定义path '*'过于宽泛，无法准确捕获motifs_loc/Ss*子目录下的生成文件。
4. 未利用Nextflow并行特性：原脚本用单进程循环处理所有文件，不符合Nextflow分布式并行的设计理念。

解决方案 #

方案1：并行化处理（推荐） #

重构管道，利用Nextflow通道自动遍历多层目录，按单个BED文件并行处理，效率更高且易维护：

#!/usr/bin/env nextflow
nextflow.enable.dsl=2

params.extract_gimme = "results/extract_gimme"
params.motifs_nr_coord = "results/motifs_nr_coord"
params.out_dir = "results/motifs_loc"

process INTERSECT_BED {
    publishDir params.out_dir, 
               mode: 'copy', 
               overwrite: false,
               saveAs: { "${sample_id}/${it.name}" }

    input:
    tuple val(sample_id), path(bed_file), path(coord_file)

    output:
    path "${bed_file.baseName}_intersected.bed"

    script:
    """
    bedtools intersect -a ${bed_file} -b ${coord_file} -wa | sort | uniq > ${bed_file.baseName}_intersected.bed
    """
}

workflow {
    // 遍历extract_gimme下所有BED文件，提取样本ID
    scan_beds = Channel.fromPath("${params.extract_gimme}/gimme_scan_*/*.bed")
        .map { file ->
            def sample_id = file.parent.name.split('_')[2]
            def bed_name = file.name
            [sample_id, bed_name, file]
        }

    // 遍历motifs_nr_coord下的BED文件，以文件名为关联键
    coord_beds = Channel.fromPath("${params.motifs_nr_coord}/*.bed")
        .map { file -> [file.name, file] }

    // 关联同名BED文件
    paired_beds = scan_beds.join(coord_beds, by: 1)
        .map { sample_id, bed_name, scan_file, coord_file ->
            [sample_id, scan_file, coord_file]
        }

    // 并行处理每对BED文件
    INTERSECT_BED(paired_beds)
}

方案2：适配原脚本的单进程处理 #

若需保留原Bash脚本逻辑，需修改脚本路径依赖并传入完整输入目录：

修改后的Bash脚本

#!/bin/bash
# 接受两个参数：extract_gimme目录路径、motifs_nr_coord目录路径
extract_gimme_dir="$1"
motifs_coord_dir="$2"

echo "intersect starts here!"
echo "------------------------"
echo " "

# 遍历extract_gimme下的样本子目录
for sample_dir in "${extract_gimme_dir}"/gimme_scan_*; do
    if [ -d "$sample_dir" ]; then
        sample_id=$(basename "$sample_dir" | cut -d'_' -f 3)
        output_dir="results/motifs_loc/${sample_id}"
        mkdir -p "$output_dir"

        # 遍历当前样本目录下的BED文件
        for bed_file in "$sample_dir"/*.bed; do
            bed_name=$(basename "$bed_file")
            coord_file="${motifs_coord_dir}/${bed_name}"
            
            if [ -f "$coord_file" ]; then
                bedtools intersect -a "$bed_file" -b "$coord_file" -wa | sort | uniq > "${output_dir}/${bed_name}_intersected.bed"
            else
                echo "Warning: 未找到配对文件 ${coord_file}，跳过 ${bed_file}"
            fi
        done

        echo "样本 ${sample_id} 处理完成!!"
        echo " "
        echo "**********************"
    fi
done

修改后的Nextflow代码

#!/usr/bin/env nextflow
nextflow.enable.dsl=2

params.extract_gimme = "results/extract_gimme"
params.motifs_nr_coord = "results/motifs_nr_coord"

process INTERSECT {
    publishDir "results", mode: 'copy', overwrite: false

    input:
    path extract_gimme_dir
    path motifs_coord_dir

    output:
    path "motifs_loc"

    script:
    """
    $baseDir/intersect_scan_coord.sh $extract_gimme_dir $motifs_coord_dir
    """
}

workflow {
    extract_gimme_ch = Channel.fromPath(params.extract_gimme, type: 'dir')
    motifs_coord_ch = Channel.fromPath(params.motifs_nr_coord, type: 'dir')
    INTERSECT(extract_gimme_ch, motifs_coord_ch)
}

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