Nextflow DSL2 入门教程¶

# 一个典型的 Shell 生信脚本
fastp -i sample_R1.fq.gz -I sample_R2.fq.gz -o clean_R1.fq.gz -O clean_R2.fq.gz
bowtie2 -x human_ref -1 clean_R1.fq.gz -2 clean_R2.fq.gz -S aligned.sam
kraken2 --db k2db --paired clean_R1.fq.gz clean_R2.fq.gz --output result.txt

# 创建一个专门的 conda 环境（避免版本冲突）
conda create -n nextflow-env -c bioconda -c conda-forge nextflow   # 从 bioconda 安装 nextflow

# 激活环境
conda activate nextflow-env   # 切换到 nextflow 环境

# 验证安装
nextflow -version   # 查看版本号，应显示 25.10.4 或更新版本

# 下载 Nextflow 启动器脚本
curl -s https://get.nextflow.io | bash   # 一行命令下载 nextflow 可执行文件

# 移动到 PATH 中（这样任何地方都能用）
sudo mv nextflow /usr/local/bin/   # 把 nextflow 移到系统路径

# 验证
nextflow -version   # 检查版本

# 如果想安装特定版本（比如 25.10.4）
export NXF_VER=25.10.4   # 设置想要的版本号
curl -s https://get.nextflow.io | bash   # 会安装指定版本

java -version   # 查看 Java 版本，需要 17 或更新

# 如果没有 Java 17，用 conda 安装
conda install -c conda-forge openjdk=17   # 安装 OpenJDK 17

// 创建一个 channel，里面放 3 个文件路径
Channel.fromPath("data/*.fastq.gz")   // 从 data 目录找所有 .fastq.gz 文件，放进水管

// 定义一个质控车间
process FASTP_QC {                          // 车间名字叫 FASTP_QC
    input:                                  // 输入：接收原始数据
    path reads                              // 一个文件路径

    output:                                 // 输出：产出清洗后的数据
    path "clean_*.fastq.gz"                 // 清洗后的文件

    script:                                 // 具体干活的命令
    """
    fastp -i ${reads} -o clean_${reads}     // 用 fastp 做质控
    """
}

workflow {                                  // 定义生产线
    reads_ch = Channel.fromPath("*.fastq.gz")   // 先把原料放进水管
    FASTP_QC(reads_ch)                      // 原料送进质控车间
    // 可以继续接更多车间...
}

Channel.fromPath("*.fastq.gz")       // 从水管取文件
    .filter { it.name =~ /sample1/ } // 阀门：只放 sample1 相关的文件通过
    .view()                          // 查看水管里有什么（调试用）

原始数据 → [Channel 水管] → [Process 车间1] → [Channel 水管] → [Process 车间2] → 最终结果
                                    ↑                                ↑
                              Operator 阀门                    Operator 阀门
                           （过滤/合并/拆分）                （过滤/合并/拆分）

                        ←———————— Workflow 生产线 ————————→

#!/usr/bin/env nextflow
// 文件名：hello.nf
// 功能：Nextflow Hello World 示例

// 定义一个 process，接收字符串并打印
process SAY_HELLO {
    input:
    val greeting                           // 输入：一个字符串值

    output:
    stdout                                 // 输出：标准输出（打印的内容）

    script:
    """
    echo "Hello, ${greeting}!"             // 打印问候语
    """
}

// 工作流：把数据送进 process
workflow {
    // 创建一个 channel，包含 3 个名字
    greetings_ch = Channel.of(             // of() 创建一个包含多个值的 channel
        "Nextflow",
        "Bioinformatics",
        "World"
    )
    // 把 channel 送进 SAY_HELLO process
    SAY_HELLO(greetings_ch)                // 调用 process（像调用函数一样）
    // 查看输出
    SAY_HELLO.out.view()                   // 打印 process 的输出结果
}

nextflow run hello.nf   # 运行 hello.nf 流程

N E X T F L O W  ~  version 25.10.4
Launching `hello.nf` [friendly_tesla] - revision: abc123
executor >  local (3)
[xx/yyyyyy] process > SAY_HELLO (1) [100%] 3 of 3 ✔

Hello, Nextflow!
Hello, Bioinformatics!
Hello, World!

conda activate bioinfo                     # 激活你的生信环境（确保有 fastp）
fastp --version                            # 验证 fastp 已安装（当前最新 v1.3.3）

mkdir -p nextflow_demo/data                # 创建项目目录
cd nextflow_demo                           # 进入项目目录
# 假设你在 data/ 目录下有一对双端测序文件：
# data/sample1_R1.fastq.gz
# data/sample1_R2.fastq.gz

#!/usr/bin/env nextflow
// 文件名：fastp_qc.nf
// 功能：用 Nextflow 调用 fastp 对单样本做质控

// ===== 参数定义 =====
params.reads_r1 = "data/sample1_R1.fastq.gz"   // 正向读段文件路径（默认值）
params.reads_r2 = "data/sample1_R2.fastq.gz"   // 反向读段文件路径（默认值）
params.outdir   = "results"                     // 输出目录（默认值）

// ===== Process 定义 =====
process FASTP {
    // publishDir：把输出文件复制到指定目录（不然只在 work/ 下）
    publishDir "${params.outdir}/fastp", mode: 'copy'   // 结果复制到 results/fastp/

    input:
    path r1                                // 输入：正向读段文件
    path r2                                // 输入：反向读段文件

    output:
    path "clean_R1.fastq.gz", emit: clean_r1     // 输出：清洗后的正向读段
    path "clean_R2.fastq.gz", emit: clean_r2     // 输出：清洗后的反向读段
    path "fastp_report.html", emit: report       // 输出：质控报告
    path "fastp_report.json", emit: json         // 输出：JSON 格式报告

    script:
    """
    fastp \\
        -i ${r1} \\
        -I ${r2} \\
        -o clean_R1.fastq.gz \\
        -O clean_R2.fastq.gz \\
        -h fastp_report.html \\
        -j fastp_report.json \\
        --thread 4 \\
        --qualified_quality_phred 20 \\
        --length_required 50
    """
    // -i/-I: 输入的正向/反向读段
    // -o/-O: 输出的正向/反向读段
    // -h/-j: HTML 和 JSON 格式的报告
    // --thread: 线程数
    // --qualified_quality_phred 20: 质量值低于 20 的碱基视为低质量
    // --length_required 50: 过滤掉长度小于 50bp 的读段
}

// ===== Workflow =====
workflow {
    // 创建输入 channel
    r1_ch = Channel.fromPath(params.reads_r1)   // 把正向读段路径放进 channel
    r2_ch = Channel.fromPath(params.reads_r2)   // 把反向读段路径放进 channel

    // 调用 FASTP process
    FASTP(r1_ch, r2_ch)                         // 把两个 channel 送进 FASTP 车间

    // 打印输出文件路径（调试用）
    FASTP.out.clean_r1.view { "Clean R1: $it" } // 查看清洗后的 R1 路径
    FASTP.out.report.view { "Report: $it" }     // 查看报告路径
}

# 基本运行
nextflow run fastp_qc.nf                       # 使用默认参数运行

# 指定自定义输入文件
nextflow run fastp_qc.nf \
    --reads_r1 data/my_R1.fq.gz \
    --reads_r2 data/my_R2.fq.gz \
    --outdir my_results                        # 用自定义参数运行

# 断点续传（之前跑过的步骤自动跳过）
nextflow run fastp_qc.nf -resume               # 加 -resume 跳过已完成的步骤

// 创建 queue channel 的几种方式
Channel.of(1, 2, 3)                        // 从值列表创建
Channel.fromPath("data/*.fastq.gz")        // 从文件路径创建
Channel.fromFilePairs("data/*_{R1,R2}.fastq.gz")  // 从配对文件创建（常用！）
Channel.fromSRA("SRP012345")               // 从 SRA 数据库创建

// 创建 value channel
Channel.value("hg38")                      // 一个字符串值
Channel.value(file("/path/to/reference.fa"))  // 一个文件路径

// 也可以用 Channel.of() 创建单值（自动变成 value channel）
ref_ch = Channel.value(file(params.reference))  // 参考基因组可以重复使用

// 场景：10 个样本都要比对到同一个参考基因组
samples_ch = Channel.fromFilePairs("data/*_{1,2}.fq.gz")  // 队列：10 个样本排队
ref_ch     = Channel.value(file("ref/hg38.fa"))            // 值：参考基因组反复用

process ALIGN {
    input:
    tuple val(sample_id), path(reads)      // 从队列取一个样本
    path reference                          // 从值通道取参考基因组（每次都能取到）

    script:
    """
    bowtie2 -x ${reference} -1 ${reads[0]} -2 ${reads[1]} -S ${sample_id}.sam
    """
}

workflow {
    ALIGN(samples_ch, ref_ch)              // 10 个样本并行比对，ref_ch 被重复使用
}

// 方法一：转义 Shell 变量
script:
"""
for line in \$(cat input.txt); do    // 注意 \$ 转义
    echo \$line
done
"""

// 方法二：用 shell 块（Nextflow 变量用 !{} 而不是 ${}）
shell:
'''
for line in $(cat input.txt); do     // Shell 变量正常写 $
    echo !{params.prefix}_$line      // Nextflow 变量用 !{}
done
'''

process MY_TASK {
    cpus 2                                 // 限制使用 2 个 CPU
    memory '4 GB'                          // 限制使用 4GB 内存

    script:
    """
    some_tool --threads ${task.cpus}       // 用 task.cpus 获取分配的 CPU 数
    """
}

# 查看出错任务的工作目录
ls work/xx/yyyyyy/                         # 看看里面有哪些文件
cat work/xx/yyyyyy/.command.log            # 查看任务日志
cat work/xx/yyyyyy/.command.err            # 查看错误信息

nextflow run ./my_pipeline.nf              # 加 ./ 明确指定本地文件

process 名字 {
    // 指令（可选）
    publishDir "输出目录", mode: 'copy'    // 发布结果文件
    cpus 4                                 // CPU 数
    memory '8 GB'                          // 内存限制
    conda 'bioconda::fastp=1.3.3'          // Conda 依赖
    container 'biocontainers/fastp:1.3.3'  // Docker 容器

    input:                                 // 输入声明
    path reads                             // 文件类型输入
    val sample_id                          // 值类型输入
    tuple val(id), path(files)             // 组合类型输入

    output:                                // 输出声明
    path "*.bam", emit: bam_files          // 文件输出（带名称）
    stdout emit: log_output                // 标准输出

    script:                                // 要执行的命令
    """
    some_command ${reads}
    """
}

// nextflow.config —— 全局配置文件
params {
    reads   = "data/*_{R1,R2}.fastq.gz"    // 默认输入路径
    outdir  = "results"                     // 默认输出路径
    threads = 4                             // 默认线程数
}

process {
    cpus   = 2                              // 所有 process 默认 2 核
    memory = '4 GB'                         // 所有 process 默认 4GB 内存
}

// 不同运行环境的配置
profiles {
    local {                                 // 本地运行
        process.executor = 'local'
    }
    slurm {                                 // SLURM 集群
        process.executor = 'slurm'
        process.queue    = 'normal'
    }
}