13. Workflow Managers & Pipeline Engines

Nextflow is a reactive dataflow programming language that decouples pipeline logic from execution environments. It allows automated task parallelization with built-in container orchestration (Docker, Singularity, Podman) and HPC/cloud schedulers (Slurm, AWS Batch, Google Cloud Life Sciences).

// 1. Nextflow DSL2 Workflow Definition (main.nf)
nextflow.enable.dsl=2

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

process FASTQC {
    tag "$sample_id"
    container 'quay.io/biocontainers/fastqc:0.12.1--hdfd78af_0'
    publishDir "${params.outdir}/fastqc", mode: 'copy'

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

    output:
    path "*_fastqc.{zip,html}", emit: qc_files

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

workflow {
    read_pairs_ch = Channel.fromFilePairs(params.reads, checkIfExists: true)
    FASTQC(read_pairs_ch)
}

// 2. Launch with Singularity profile on local cluster
// nextflow run main.nf -profile singularity --outdir ./results -resume

nf-core is a community-driven initiative providing peer-reviewed, highly standardized, production-ready NGS pipelines built on Nextflow. Popular pipelines include nf-core/rnaseq, nf-core/sarek (germline & somatic variant calling), and nf-core/ampliseq.

# 1. Install nf-core CLI helper tools
pip install nf-core

# 2. Launch production RNA-Seq pipeline with Singularity container profile
nextflow run nf-core/rnaseq \
  -r 3.14.0 \
  -profile singularity \
  --input samplesheet.csv \
  --genome GRCh38 \
  --aligner star_salmon \
  --outdir ./rnaseq_results/ \
  -resume

# 3. Launch somatic cancer calling with nf-core/sarek
# nextflow run nf-core/sarek -r 3.4.2 -profile singularity --input cancer_samples.csv --genome GRCh38

Snakemake is a workflow management engine based on Python. It describes workflows via human-readable declarative rules, tracking input/output file dependencies and seamlessly managing isolated Conda software environments and Slurm submissions.

# 1. Define Snakefile with automated conda environments
SAMPLES = ["sample1", "sample2"]

rule all:
    input:
        expand("bams/{sample}.sorted.bam.bai", sample=SAMPLES)

rule bwa_mem:
    input:
        r1="fastqs/{sample}_R1.fastq.gz",
        r2="fastqs/{sample}_R2.fastq.gz",
        ref="ref/genome.fa"
    output:
        temp("bams/{sample}.raw.bam")
    threads: 8
    conda:
        "envs/bwa.yaml"
    shell:
        "bwa mem -t {threads} {input.ref} {input.r1} {input.r2} | samtools view -bS - > {output}"

rule samtools_sort:
    input:
        "bams/{sample}.raw.bam"
    output:
        bam="bams/{sample}.sorted.bam",
        bai="bams/{sample}.sorted.bam.bai"
    threads: 4
    shell:
        "samtools sort -@ {threads} -o {output.bam} {input} && samtools index {output.bam}"

# 2. Execute Snakemake workflow with 16 cores and Conda:
# snakemake --cores 16 --use-conda --rerun-incomplete

Workflow Description Language (WDL) is a domain-specific language designed by the Broad Institute to specify data-processing workflows with human-readable syntax. It powers the official GATK Best Practices on Cromwell execution engines and Terra.bio.

version 1.0

task HaplotypeCaller {
  input {
    File input_bam
    File input_bam_index
    File ref_fasta
    File ref_fasta_index
    String sample_name
  }
  command <<<
    gatk --java-options "-Xmx8g" HaplotypeCaller \
      -R ~{ref_fasta} \
      -I ~{input_bam} \
      -O ~{sample_name}.g.vcf.gz \
      -ERC GVCF
  >>>
  runtime {
    docker: "broadinstitute/gatk:4.5.0.0"
    cpu: 4
    memory: "16 GB"
  }
  output {
    File gvcf = "~{sample_name}.g.vcf.gz"
  }
}

workflow GermlineCalling {
  input {
    File bam
    File bai
    File ref
    File ref_idx
    String sample
  }
  call HaplotypeCaller {
    input:
      input_bam = bam, input_bam_index = bai,
      ref_fasta = ref, ref_fasta_index = ref_idx,
      sample_name = sample
  }
}

Galaxy is an open, web-based platform for accessible, reproducible, and transparent computational biology. For power users and automated production, the BioBlend Python API connects to Galaxy servers to run complex histories and workflows programmatically.

from bioblend.galaxy import GalaxyInstance

# 1. Connect to Galaxy instance via API key
GALAXY_URL = "https://usegalaxy.org"
API_KEY = "your_galaxy_api_key_here"
gi = GalaxyInstance(url=GALAXY_URL, key=API_KEY)

# 2. Create a new analysis history
new_hist = gi.histories.create_history(name="Automated_NGS_QC_Workflow")
history_id = new_hist["id"]

# 3. Upload paired-end FASTQs to history
file_r1 = gi.tools.upload_file("sample_R1.fq.gz", history_id, file_type="fastqsanger.gz")
file_r2 = gi.tools.upload_file("sample_R2.fq.gz", history_id, file_type="fastqsanger.gz")

# 4. Invoke pre-configured FastQC + MultiQC workflow
wf = gi.workflows.get_workflows(name="Standard_Illumina_QC")[0]
invocation = gi.workflows.invoke_workflow(
    workflow_id=wf["id"],
    history_id=history_id,
    inputs={"0": {"id": file_r1["outputs"][0]["id"], "src": "hda"}}
)
print("Workflow successfully launched with Invocation ID:", invocation["id"])