Nextflow脚本处理给定目录下的所有文件

Question

我有一个nextflow脚本，它在单个vcf文件上运行几个进程。文件名为"bos_taurus"。它位于目录/input_files/bos_taurus.vcf中。目录input_files/还包含另一个文件'sacharomyces_cerevisea.vcf'。我想我的nextflow脚本处理这两个文件。我试图使用像ch_1 = channel.fromPath("/input_files/*.vcf")这样的glob模式，但遗憾的是我找不到一个有效的解决方案。如有任何帮助，我将不胜感激。

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

// here I tried to use globbing
params.input_files = "/mnt/c/Users/Lenovo/Desktop/STUDIA/BIOINFORMATYKA/SEMESTR_V/PRACOWNIA_INFORMATYCZNA/nextflow/projekt/input_files/*.vcf"
params.results_dir = "/mnt/c/Users/Lenovo/Desktop/STUDIA/BIOINFORMATYKA/SEMESTR_V/PRACOWNIA_INFORMATYCZNA/nextflow/projekt/results"

file_channel = Channel.fromPath( params.input_files, checkIfExists: true )

// how can I make this process work on two files simultanously
process FILTERING {
publishDir("${params.results_dir}/after_filtering", mode: 'copy')
input:
path(input_files)
output:
path("*")
script:
"""
vcftools --vcf ${input_files} --mac 1 --minQ 20 --recode  --recode-INFO-all  --out after_filtering.vcf
"""
}

Answer 1

注意，如果您的VCF文件实际上是bgzip压缩和tabix索引的，那么您可以使用fromFilePairs工厂方法来创建您的输入通道。例如:

params.vcf_files = "./input_files/*.vcf.gz{,.tbi}"
params.results_dir = "./results"

process FILTERING {
tag { sample }
publishDir("${params.results_dir}/after_filtering", mode: 'copy')
input:
tuple val(sample), path(indexed_vcf)
output:
tuple val(sample), path("${sample}.filtered.vcf")
"""
vcftools \
--vcf "${indexed_vcf.first()}" \
--mac 1 \
--minQ 20 \
--recode \
--recode-INFO-all \
--out "${sample}.filtered.vcf"
"""
}
workflow {
vcf_files = Channel.fromFilePairs( params.vcf_files, checkIfExists: true )
FILTERING( vcf_files ).view()
}

结果:

$ nextflow run main.nf
N E X T F L O W  ~  version 22.10.0
Launching `main.nf` [thirsty_torricelli] DSL2 - revision: 8f69ad5638
executor >  local (3)
[7d/dacad6] process > FILTERING (C) [100%] 3 of 3 ✔
[A, /path/to/work/84/f9f00097bcd2b012d3a5e105b9d828/A.filtered.vcf]
[B, /path/to/work/cb/9f6f78213f0943013990d30dbb9337/B.filtered.vcf]
[C, /path/to/work/7d/dacad693f06025a6301c33fd03157b/C.filtered.vcf]

请注意，BCFtools是积极维护的，并打算作为VCFtools的替代品。在生产管道中，应该优先使用BCFtools。

Answer 2

这里有一个初学者的小示例。首先，应该在每个流程中指定唯一的输出名称。目前，after_filtering.vcf是硬编码的，因此一旦复制到publishDir，它们将相互覆盖。您可以使用baseName操作符，如下所示，并将其永久存储在输入文件通道中，第一个元素是示例名称，第二个元素是实际文件。我做了一个例子过程，只是运行head上的vcf，然后你可以根据你实际需要的调整。

#! /usr/bin/env nextflow
nextflow.enable.dsl = 2
params.input_files = "/Users/atpoint/vcf/*.vcf"
params.results_dir = "/Users/atpoint/vcf/"
// A channel that contains a map with sample name and the file itself
file_channel = Channel.fromPath( params.input_files, checkIfExists: true )
.map { it -> [it.baseName, it] }
// An example process just head-ing the vcf
process VcfHead {
publishDir("${params.results_dir}/after_filtering", mode: 'copy')
input:
tuple val(name), path(vcf_in)
output:
path("*_head.vcf")
script:
""" 
head -n 1 $vcf_in > ${name}_head.vcf
"""
}                      
// Run it
workflow {
VcfHead(file_channel)
}

file_channel通道看起来像这样，如果你添加一个.view():

[one, /Users/atpoint/vcf/one.vcf]
[two, /Users/atpoint/vcf/two.vcf]