nextflow-workflow-engine

Nextflow — 可扩展的科学工作流引擎

概述

Nextflow 实现了一种数据流编程模型，其中 流程（容器化的执行单元）通过 通道（异步队列）接收和发送数据。这种设计可实现隐式并行化——只要输入通道中有数据，流程就会立即运行，无需手动管理依赖关系。Nextflow 通过切换单个配置配置文件，即可在本地计算机、HPC 集群（SLURM、SGE、PBS）和云平台（AWS Batch、Google Cloud Life Sciences、Azure Batch）上编排流程。nf-core 社区提供了 100 多条经过验证的 Nextflow 工作流（RNA-seq、WGS、ChIP-seq、scRNA-seq），遵循最佳实践并配有自动化测试。

适用场景

• 构建容器化生物信息学工作流，无需修改代码即可在 HPC、AWS 和本地环境中运行
• 开箱即用地使用 nf-core 社区工作流（nf-core/rnaseq、nf-core/sarek、nf-core/chipseq）
• 在 SLURM 集群上通过隐式并行化处理数千个样本
• 编写工作流，让每个步骤都在 Docker 或 Singularity 容器内运行，以确保可复现性
• 监控工作流执行情况，并在故障后使用 -resume 从检查点恢复
• 如果优先考虑 Python 集成，则改用 Snakemake，以构建原生 Python 的基于规则的工作流
• 如果临床基因组学工作流需要符合 CWL/WDL 标准，则改用 WDL/Cromwell

前置条件

• 软件：Java 11+、Nextflow（自包含启动器）
• 容器：用于隔离流程的 Docker 或 Singularity（推荐）
• 可选：用于管理社区工作流的 nf-core tools

安装前请检查：当前环境中可能已提供该工具（例如，在 pixi / conda 环境中）。先运行 command -v nextflow；如果返回路径，则跳过下方的安装命令。在 pixi 项目中运行时，请通过 pixi run nextflow 调用该工具，而不要直接运行 nextflow。

# Install Nextflow (self-contained JAR — no sudo required)
curl -s https://get.nextflow.io | bash
chmod +x nextflow
export PATH="$PWD:$PATH"

# Verify
nextflow -version
# Nextflow version 24.10.1

# Install nf-core tools (Python)
pip install nf-core

# Pull an nf-core pipeline
nextflow pull nf-core/rnaseq

快速开始

// hello.nf — minimal Nextflow pipeline
nextflow.enable.dsl = 2

process GREET {
    input: val name
    output: stdout
    script: "echo 'Hello, ${name}!'"
}

workflow {
    Channel.of('World', 'Nextflow') | GREET | view
}

# Run the pipeline
nextflow run hello.nf
# Hello, World!
# Hello, Nextflow!

核心 API

模块 1：流程——容器化任务单元

使用输入、输出和 shell/script 指令定义流程。

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

process ALIGN_READS {
    // Container for this process
    container 'quay.io/biocontainers/star:2.7.11a--h0033a41_0'
    
    // Resource directives
    cpus 16
    memory '32 GB'
    
    // I/O declarations
    input:
    tuple val(sample_id), path(reads_r1), path(reads_r2)
    path genome_index
    
    output:
    tuple val(sample_id), path("${sample_id}.Aligned.sortedByCoord.out.bam")
    path "${sample_id}.Log.final.out", emit: log
    
    // Shell command
    script:
    """
    STAR --runThreadN ${task.cpus} \\
         --genomeDir ${genome_index} \\
         --readFilesIn ${reads_r1} ${reads_r2} \\
         --readFilesCommand zcat \\
         --outSAMtype BAM SortedByCoordinate \\
         --outFileNamePrefix ${sample_id}.
    """
}

模块 2：通道——流程之间的数据队列

创建并转换通道，实现灵活的数据路由。

nextflow.enable.dsl = 2

workflow {
    // Value channel (broadcast)
    genome_ch = Channel.value(file("GRCh38.fa"))
    
    // List channel
    samples_ch = Channel.of('ctrl_1', 'ctrl_2', 'treat_1', 'treat_2')
    
    // File channel from glob pattern
    reads_ch = Channel.fromFilePairs("data/*_{R1,R2}.fastq.gz")
    // Emits: [sample_id, [R1_file, R2_file]]
    
    // From a CSV sample sheet
    sample_sheet = Channel.fromPath("samplesheet.csv")
        .splitCsv(header: true)
        .map { row -> tuple(row.sample, file(row.fastq_1), file(row.fastq_2)) }
    
    // Channel operators
    filtered = reads_ch
        .filter { id, files -> id.startsWith("ctrl") }
        .view { id, files -> "Processing: ${id}" }
}

模块 3：工作流块——管线 DAG 定义

通过通道连接流程，以定义管线 DAG。

nextflow.enable.dsl = 2

include { FASTP } from './modules/fastp'
include { STAR_ALIGN } from './modules/star'
include { FEATURECOUNTS } from './modules/featurecounts'

workflow RNA_SEQ {
    take:
    reads_ch     // tuple: [sample_id, [R1, R2]]
    genome_idx   // path: STAR index directory
    gtf          // path: annotation GTF file
    
    main:
    // Trim reads
    FASTP(reads_ch)
    
    // Align trimmed reads
    STAR_ALIGN(FASTP.out.reads, genome_idx)
    
    // Count reads (join on sample_id)
    FEATURECOUNTS(STAR_ALIGN.out.bam, gtf)
    
    emit:
    counts = FEATURECOUNTS.out.counts
    logs   = STAR_ALIGN.out.log.mix(FASTP.out.log)
}

workflow {
    reads = Channel.fromFilePairs("data/*_{R1,R2}.fastq.gz")
    genome_idx = Channel.value(file("genome/star_index"))
    gtf = Channel.value(file("genome/annotation.gtf"))
    
    RNA_SEQ(reads, genome_idx, gtf)
    RNA_SEQ.out.counts | view
}

模块 4：配置——适用于不同环境的配置文件

配置适用于本地、HPC 和云环境的执行配置文件。

// nextflow.config — profile-based configuration
profiles {
    local {
        process.executor = 'local'
        process.cpus = 4
        process.memory = '8 GB'
        docker.enabled = true
    }
    
    slurm {
        process.executor = 'slurm'
        process.queue = 'batch'
        process.clusterOptions = '--account=myproject'
        singularity.enabled = true
        singularity.autoMounts = true
        
        // Per-process resource configuration
        process {
            withName: STAR_ALIGN {
                cpus = 16
                memory = '32 GB'
                time = '4h'
            }
            withName: FASTP {
                cpus = 8
                memory = '8 GB'
            }
        }
    }
    
    aws {
        process.executor = 'awsbatch'
        process.queue = 'arn:aws:batch:us-east-1:123456789:job-queue/my-queue'
        aws.region = 'us-east-1'
        aws.batch.cliPath = '/home/ec2-user/miniconda/bin/aws'
        docker.enabled = true
    }
}

// Global params
params {
    outdir    = 'results'
    genome    = 'GRCh38'
    max_cpus  = 16
    max_memory = '128 GB'
    max_time   = '72h'
}

模块 5：操作符——通道转换

使用内置操作符转换通道。

nextflow.enable.dsl = 2

workflow {
    // Map: transform each element
    Channel.fromPath("data/*.fastq.gz")
        .map { file -> tuple(file.baseName.replaceAll(/_R[12]/, ''), file) }
        .groupTuple()   // group by sample_id → [sample_id, [R1, R2]]
        .view { id, files -> "Sample: ${id} (${files.size()} files)" }
    
    // Filter by condition
    Channel.of(1, 2, 3, 4, 5)
        .filter { it > 3 }
        .view()  // emits: 4, 5
    
    // Combine channels
    samples = Channel.of('A', 'B', 'C')
    refs = Channel.value(file("genome.fa"))
    samples.combine(refs).view()  // [A, genome.fa], [B, genome.fa], [C, genome.fa]
    
    // Collect all outputs into a list
    Channel.of(1, 2, 3).collect().view()  // [[1, 2, 3]]
}

模块 6：错误处理与恢复

处理故障、重试策略和工作流恢复。

// nextflow.config — retry and error handling
process {
    // Retry failed processes up to 3 times with increasing memory
    errorStrategy = { task.exitStatus in [137, 140] ? 'retry' : 'finish' }
    maxRetries = 3
    memory = { 8.GB * task.attempt }   // 8GB → 16GB → 24GB on retries
    
    // Ignore errors for optional steps
    withName: OPTIONAL_STEP {
        errorStrategy = 'ignore'
    }
}

# Resume a pipeline from the last successful checkpoint
nextflow run pipeline.nf -resume

# View process execution statistics
nextflow log amazing_fermi  # use run name from .nextflow.log

# Inspect work directories
ls work/ab/cdef1234*/

关键参数

参数	默认值	范围/选项	作用
-resume	关闭	标志	使用缓存结果从上次成功的检查点恢复
-profile	—	local、slurm、aws、自定义	从 nextflow.config 中选择执行配置文件
-params-file	—	JSON/YAML 路径	从文件加载管线参数
-w / --work-dir	./work	目录路径	中间文件的工作目录
process.cpus	1	整数	每个流程的默认 CPU 数量；可使用 withName 覆盖
process.memory	1 GB	内存字符串	每个流程的默认内存
process.executor	local	local、slurm、sge、awsbatch、k8s	作业调度器或云执行器
process.errorStrategy	terminate	retry、ignore、finish	流程失败时的处理方式
process.maxRetries	0	整数	失败前自动重试的最大次数
docker.enabled	false	布尔值	启用 Docker 容器运行时

常见工作流

工作流 1：完整的 RNA-seq 工作流（nf-core/rnaseq）

# Run nf-core/rnaseq with a samplesheet
# samplesheet.csv: sample,fastq_1,fastq_2,strandedness
cat > samplesheet.csv << 'EOF'
sample,fastq_1,fastq_2,strandedness
ctrl_1,data/ctrl_1_R1.fastq.gz,data/ctrl_1_R2.fastq.gz,auto
ctrl_2,data/ctrl_2_R1.fastq.gz,data/ctrl_2_R2.fastq.gz,auto
treat_1,data/treat_1_R1.fastq.gz,data/treat_1_R2.fastq.gz,auto
EOF

# Run nf-core/rnaseq (downloads pipeline and containers automatically)
nextflow run nf-core/rnaseq \
    -profile docker \
    --input samplesheet.csv \
    --genome GRCh38 \
    --outdir results/ \
    -resume

echo "Results in: results/star_salmon/  results/multiqc/"

工作流 2：使用子工作流的自定义模块化流水线

// main.nf — modular pipeline structure
nextflow.enable.dsl = 2

include { QC_TRIM     } from './subworkflows/qc_trim'
include { ALIGN_COUNT } from './subworkflows/align_count'
include { MULTIQC     } from './modules/multiqc'

workflow {
    // Input: sample sheet CSV
    ch_samples = Channel
        .fromPath(params.samplesheet)
        .splitCsv(header: true)
        .map { row -> tuple(row.sample, file(row.fastq_1), file(row.fastq_2)) }
    
    // QC and trimming
    QC_TRIM(ch_samples)
    
    // Alignment and counting
    ALIGN_COUNT(
        QC_TRIM.out.reads,
        file(params.genome_index),
        file(params.gtf)
    )
    
    // Aggregate QC
    all_logs = QC_TRIM.out.logs.mix(ALIGN_COUNT.out.logs).collect()
    MULTIQC(all_logs)
    
    // Publish count matrix
    ALIGN_COUNT.out.counts.view { id, file ->
        "Counts: ${id} → ${file}"
    }
}

常见用法

用法 1：监控运行中的流水线并查看日志

# View current pipeline run status
nextflow log

# Detailed log for a specific run
nextflow log amazing_fermi -f name,status,exit,duration,realtime,rss

# Watch pipeline progress in real-time
tail -f .nextflow.log

# Generate HTML execution report and timeline
nextflow run pipeline.nf -with-report report.html -with-timeline timeline.html
open report.html

用法 2：使用自定义配置运行 nf-core 流水线

# Create custom config for institutional HPC
cat > custom.config << 'EOF'
process {
    executor = 'slurm'
    queue = 'normal'
    clusterOptions = '--account=myproject'
    
    withName: STAR_ALIGN {
        cpus = 16
        memory = '32 GB'
        time = '4h'
    }
}

singularity {
    enabled = true
    autoMounts = true
    cacheDir = '/scratch/singularity_cache'
}
EOF

nextflow run nf-core/rnaseq \
    -profile singularity \
    -c custom.config \
    --input samplesheet.csv \
    --genome GRCh38 \
    --outdir results/ \
    -resume

故障排除

问题	原因	解决方案
Cannot find a matching process	未包含 DSL2 模块	添加 include { PROCESS_NAME } from './modules/...'
工作目录占满磁盘	中间文件不断累积	运行 nextflow clean -f 删除旧的工作目录
容器拉取失败	注册表身份验证或网络问题	预先拉取：docker pull quay.io/biocontainers/tool:version；使用 --pull never
resume 未跳过已完成的步骤	流程输入已更改或缓存失效	检查输入文件或参数是否已更改；使用 nextflow log 检查缓存
SLURM 作业一直处于等待状态	队列资源不足	使用 squeue 检查；在配置中减少 memory 和 cpus
Java 中出现 OutOfMemoryError	Nextflow JVM 堆太小	export NXF_JVM_ARGS="-Xms1g -Xmx8g"
未找到流程输出	output: 块中的输出路径错误	检查工作目录：ls work/xx/yyyyyy*/；确认脚本已生成预期文件
nf-core 流水线校验和失败	流水线缓存过期	运行 nextflow pull nf-core/rnaseq -r latest 更新

参考资料

• Nextflow 文档 — 官方语言参考和执行器指南
• Nextflow GitHub：nextflow-io/nextflow — 源代码和发行版本
• Di Tommaso P 等（2017）“Nextflow enables reproducible computational workflows” — Nature Biotechnology 35:316-319。DOI:10.1038/nbt.3820
• nf-core — 社区策划的 Nextflow 流水线，提供自动化测试和容器