Nextflow Pipelines

Basic Pipeline Structure

// main.nf nextflow.enable.dsl=2

params.reads = "data/*_{1,2}.fq.gz" params.outdir = "results"

process FASTQC { input: tuple val(sample_id), path(reads)

output:
path("*.html"), emit: html
path("*.zip"), emit: zip

script:
"""
fastqc ${reads}
"""

}

workflow { Channel.fromFilePairs(params.reads) | FASTQC }

DSL2 Modules

// modules/fastqc.nf process FASTQC { tag "${sample_id}" publishDir "${params.outdir}/qc", mode: 'copy'

input:
tuple val(sample_id), path(reads)

output:
tuple val(sample_id), path("*.html"), emit: html
tuple val(sample_id), path("*.zip"), emit: zip

script:
"""
fastqc -t ${task.cpus} ${reads}
"""

}

// main.nf include { FASTQC } from './modules/fastqc' include { ALIGN } from './modules/align'

workflow { reads_ch = Channel.fromFilePairs(params.reads) FASTQC(reads_ch) ALIGN(reads_ch) }

Config File

// nextflow.config params { reads = "data/*_{1,2}.fq.gz" outdir = "results" genome = "ref/genome.fa" }

process { cpus = 4 memory = '8 GB' time = '2h'

withName: 'ALIGN' {
    cpus = 16
    memory = '32 GB'
}

}

profiles { docker { docker.enabled = true } singularity { singularity.enabled = true } slurm { process.executor = 'slurm' } }

Container Support

process SALMON_QUANT { container 'quay.io/biocontainers/salmon:1.10.0--h7e5ed60_0'

input:
tuple val(sample_id), path(reads)
path(index)

output:
tuple val(sample_id), path("${sample_id}"), emit: quant

script:
"""
salmon quant -i ${index} -l A -1 ${reads[0]} -2 ${reads[1]} \
    -o ${sample_id} --threads ${task.cpus}
"""

}

Channel Operations

// From file pairs Channel.fromFilePairs("data/*_{1,2}.fq.gz") .set { reads_ch }

// From path Channel.fromPath("data/*.bam") .map { file -> tuple(file.baseName, file) } .set { bam_ch }

// From samplesheet Channel.fromPath(params.samplesheet) .splitCsv(header: true) .map { row -> tuple(row.sample, file(row.fastq_1), file(row.fastq_2)) } .set { samples_ch }

// Combine channels reads_ch.combine(reference_ch)

Subworkflows

// subworkflows/qc.nf include { FASTQC } from '../modules/fastqc' include { MULTIQC } from '../modules/multiqc'

workflow QC { take: reads

main:
FASTQC(reads)
MULTIQC(FASTQC.out.zip.collect())

emit:
qc_report = MULTIQC.out.report

}

// main.nf include { QC } from './subworkflows/qc' include { ALIGN } from './subworkflows/align'

workflow { reads = Channel.fromFilePairs(params.reads) QC(reads) ALIGN(reads) }

Cluster Execution

// nextflow.config for SLURM process { executor = 'slurm' queue = 'normal' clusterOptions = '--account=myproject'

withLabel: 'high_memory' {
    memory = '128 GB'
    queue = 'highmem'
}

}

executor { name = 'slurm' queueSize = 100 submitRateLimit = '10 sec' }

AWS/Cloud Execution

// nextflow.config for AWS Batch process { executor = 'awsbatch' queue = 'my-batch-queue' }

aws { region = 'us-east-1' batch { cliPath = '/usr/local/bin/aws' } }

Run on AWS

nextflow run main.nf -profile awsbatch -bucket-dir s3://my-bucket/work

Resource Labels

process { withLabel: 'process_low' { cpus = 2 memory = '4 GB' time = '1h' } withLabel: 'process_medium' { cpus = 8 memory = '16 GB' time = '4h' } withLabel: 'process_high' { cpus = 16 memory = '64 GB' time = '12h' } }

process ALIGN { label 'process_high' // ... }

Error Handling

process RISKY_PROCESS { errorStrategy 'retry' maxRetries 3 memory { 8.GB * task.attempt }

script:
"""
memory_intensive_command
"""

}

process OPTIONAL_PROCESS { errorStrategy 'ignore' // ... }

Caching and Resume

Resume from last run

nextflow run main.nf -resume

Clean work directory

nextflow clean -f

Show execution trace

nextflow log

Complete RNA-seq Pipeline

nextflow.enable.dsl=2

params.reads = "data/*_{1,2}.fq.gz" params.salmon_index = "ref/salmon_index" params.outdir = "results"

process FASTP { tag "${sample_id}" publishDir "${params.outdir}/trimmed", mode: 'copy'

input:
tuple val(sample_id), path(reads)

output:
tuple val(sample_id), path("${sample_id}_{1,2}.trimmed.fq.gz"), emit: reads
path("${sample_id}.json"), emit: json

script:
"""
fastp -i ${reads[0]} -I ${reads[1]} \
    -o ${sample_id}_1.trimmed.fq.gz -O ${sample_id}_2.trimmed.fq.gz \
    --json ${sample_id}.json --thread ${task.cpus}
"""

}

process SALMON_QUANT { tag "${sample_id}" publishDir "${params.outdir}/salmon", mode: 'copy'

input:
tuple val(sample_id), path(reads)
path(index)

output:
tuple val(sample_id), path("${sample_id}"), emit: quant

script:
"""
salmon quant -i ${index} -l A -1 ${reads[0]} -2 ${reads[1]} \
    -o ${sample_id} --threads ${task.cpus}
"""

}

process MULTIQC { publishDir "${params.outdir}", mode: 'copy'

input:
path('*')

output:
path("multiqc_report.html")

script:
"""
multiqc .
"""

}

workflow { reads_ch = Channel.fromFilePairs(params.reads) index_ch = Channel.fromPath(params.salmon_index)

FASTP(reads_ch)
SALMON_QUANT(FASTP.out.reads, index_ch.first())

qc_files = FASTP.out.json.collect()
    .mix(SALMON_QUANT.out.quant.collect())
MULTIQC(qc_files.collect())

}

Related Skills

• workflow-management/snakemake-workflows - Snakemake alternative

• workflows/rnaseq-to-de - End-to-end RNA-seq

• read-qc/fastp-workflow - QC processes