Nextflow 配置与多平台部署¶

// nextflow.config — params 区块
// 定义流程运行时的各种参数，用户可以通过命令行覆盖

params {
    // === 输入输出 ===
    reads       = "data/*_R{1,2}.fastq.gz"         // 原始测序数据路径（支持通配符）
    outdir      = "results"                         // 结果输出目录
    samplesheet = null                              // 样本信息表（可选）

    // === 参考数据 ===
    host_index  = "/db/bowtie2/GRCh38"              // 宿主基因组索引路径
    kraken2_db  = "/db/kraken2/standard"            // Kraken2 数据库路径

    // === 工具参数 ===
    fastp_qualified_quality = 20                    // fastp 碱基质量阈值
    fastp_min_length        = 50                    // fastp 最短读长

    // === 流程控制 ===
    skip_dehost = false                             // 是否跳过去宿主步骤
    max_cpus    = 16                                // 最大 CPU 数上限
    max_memory  = '64.GB'                           // 最大内存上限
    max_time    = '24.h'                            // 最大运行时间上限
}

# 用双横杠 -- 覆盖 params 里的任何参数
nextflow run main.nf \
    --reads "new_data/*_R{1,2}.fastq.gz" \         # 覆盖输入路径
    --outdir "new_results" \                        # 覆盖输出目录
    --fastp_qualified_quality 30                    # 覆盖质量阈值

// nextflow.config — process 区块
// 给每个 process 分配 CPU、内存、运行时间等资源

process {
    // === 默认资源（所有 process 的兜底配置）===
    cpus   = 1                                      // 默认 1 个 CPU
    memory = '2.GB'                                 // 默认 2GB 内存
    time   = '1.h'                                  // 默认 1 小时超时

    // === 按标签分配资源 ===
    // 模块里写 label 'process_low' 的 process 用这套资源
    withLabel: 'process_low' {
        cpus   = 2                                  // 2 个 CPU
        memory = '4.GB'                             // 4GB 内存
        time   = '2.h'                              // 2 小时
    }

    withLabel: 'process_medium' {
        cpus   = 4                                  // 4 个 CPU
        memory = '8.GB'                             // 8GB 内存
        time   = '4.h'                              // 4 小时
    }

    withLabel: 'process_high' {
        cpus   = 8                                  // 8 个 CPU
        memory = '32.GB'                            // 32GB 内存
        time   = '8.h'                              // 8 小时
    }

    // === 按 process 名字分配资源 ===
    // 更精确的控制：只给特定 process 配资源
    withName: 'KRAKEN2' {
        cpus   = 8                                  // Kraken2 需要更多 CPU
        memory = '48.GB'                            // Kraken2 数据库很大，需要大内存
        time   = '12.h'                             // 大数据量可能跑很久
    }

    withName: 'FASTP' {
        cpus   = 4                                  // fastp 4 核就够了
        memory = '4.GB'                             // fastp 不吃内存
    }
}

// nextflow.config — profiles 区块
// 不同环境的配置方案，通过 -profile 切换

profiles {
    // 本地运行（默认）
    standard {
        process.executor = 'local'                  // 在本地执行
        docker.enabled   = false                    // 不用 Docker
    }

    // Docker 容器
    docker {
        docker.enabled   = true                     // 启用 Docker
        docker.runOptions = '-u $(id -u):$(id -g)'  // 保持文件权限一致
    }

    // HPC 集群（SLURM）
    slurm {
        process.executor = 'slurm'                  // 使用 SLURM 调度
        process.queue    = 'normal'                 // 默认队列
        singularity.enabled = true                  // HPC 上通常用 Singularity
    }

    // 测试 profile（用小数据快速验证流程）
    test {
        params.reads      = "test_data/*_R{1,2}.fastq.gz"
        params.max_cpus   = 2
        params.max_memory = '4.GB'
        params.max_time   = '30.m'
    }
}

# 用 -profile 参数选择运行环境
nextflow run main.nf -profile docker               # 用 Docker 在本地跑
nextflow run main.nf -profile slurm                # 在 SLURM 集群上跑
nextflow run main.nf -profile test,docker           # 组合使用：测试数据 + Docker

// nextflow.config — 完整实战配置

// === 参数定义 ===
params {
    reads       = "data/*_R{1,2}.fastq.gz"
    outdir      = "results"
    host_index  = "/db/bowtie2/GRCh38"
    kraken2_db  = "/db/kraken2/standard"
    max_cpus    = 16
    max_memory  = '64.GB'
    max_time    = '48.h'
}

// === 默认 process 资源 ===
process {
    cpus   = { 1 * task.attempt }                   // 重试时自动翻倍
    memory = { 2.GB * task.attempt }                // 重试时自动加内存
    time   = { 1.h * task.attempt }                 // 重试时自动加时间

    errorStrategy = 'retry'                         // 失败时重试
    maxRetries    = 2                               // 最多重试 2 次

    withLabel: 'process_high' {
        cpus   = { 8 * task.attempt }
        memory = { 32.GB * task.attempt }
        time   = { 8.h * task.attempt }
    }
}

// === Profile 定义 ===
profiles {

    // ===== Profile 1: 本地笔记本 =====
    local {
        process.executor = 'local'                  // 本地执行，不提交到调度器
        params.max_cpus   = 4                       // 笔记本一般就 4 核
        params.max_memory = '8.GB'                  // 内存限制 8GB
        params.max_time   = '4.h'                   // 最长跑 4 小时
        docker.enabled    = false                   // 本地可能没装 Docker
    }

    // ===== Profile 2: SLURM 集群 =====
    slurm {
        process {
            executor       = 'slurm'                // SLURM 调度器
            queue          = 'normal'               // 提交到 normal 队列
            clusterOptions = '--account=mylab'      // 指定账户（按你的集群改）
        }
        executor {
            queueSize       = 50                    // 最多同时提交 50 个作业
            pollInterval    = '1 min'               // 每分钟检查一次作业状态
            submitRateLimit = '10/1min'              // 每分钟最多提交 10 个
        }
        params.max_cpus   = 32                      // 集群节点通常有 32 核
        params.max_memory = '128.GB'                // 集群节点通常有 128GB
        params.max_time   = '48.h'                  // 最长 48 小时
    }

    // ===== Profile 3: PBS/Torque 集群 =====
    pbs {
        process {
            executor = 'pbs'                        // PBS 调度器
            queue    = 'batch'                      // 提交到 batch 队列
        }
        executor {
            queueSize = 30                          // 最多同时 30 个作业
        }
    }

    // ===== Profile 4: Docker 容器 =====
    docker {
        docker.enabled    = true                    // 启用 Docker
        docker.runOptions = '-u $(id -u):$(id -g)'  // 文件权限：用当前用户身份运行
        // 每个 process 在模块文件里用 container 指令指定镜像
        // 不需要在这里统一指定
    }

    // ===== Profile 5: Singularity（HPC 常用）=====
    singularity {
        singularity.enabled    = true               // 启用 Singularity
        singularity.autoMounts = true               // 自动挂载主机目录
        singularity.cacheDir   = "${HOME}/.singularity/cache"  // 镜像缓存目录
        // Singularity 可以直接拉 Docker 镜像
        // 模块里写的 container "quay.io/..." 会自动转成 Singularity 格式
    }

    // ===== Profile 6: AWS Batch（云端）=====
    awsbatch {
        process {
            executor  = 'awsbatch'                  // AWS Batch 执行器
            queue     = 'my-batch-queue'             // AWS Batch 队列名
            container = 'quay.io/biocontainers/fastp:0.24.0--heae7151_0'  // 默认容器
        }
        aws {
            region = 'us-east-1'                    // AWS 区域
            batch {
                cliPath = '/home/ec2-user/miniconda/bin/aws'  // AWS CLI 路径
            }
        }
        params.outdir = "s3://my-bucket/results"    // 结果存到 S3
        workDir       = "s3://my-bucket/work"       // 工作目录也在 S3
    }

    // ===== 组合 profile 示例 =====
    // SLURM + Singularity（最常见的 HPC 组合）
    hpc {
        process {
            executor = 'slurm'
            queue    = 'normal'
        }
        singularity.enabled    = true
        singularity.autoMounts = true
    }

    // 测试 profile（小数据快速验证）
    test {
        params.reads      = "${projectDir}/test_data/*_R{1,2}.fastq.gz"
        params.max_cpus   = 2
        params.max_memory = '4.GB'
        params.max_time   = '30.m'
    }
}

# 在笔记本上用小数据测试
nextflow run main.nf -profile test,local

# 在 HPC 上用 SLURM + Singularity 跑正式数据
nextflow run main.nf -profile slurm,singularity

# 在 AWS 上跑
nextflow run main.nf -profile awsbatch

# 本地用 Docker 跑
nextflow run main.nf -profile docker

// nextflow.config — 动态资源
process {
    // task.attempt 是当前第几次尝试（从 1 开始）
    cpus   = { 2 * task.attempt }                   // 第 1 次 2 核，第 2 次 4 核，第 3 次 6 核
    memory = { 4.GB * task.attempt }                // 第 1 次 4GB，第 2 次 8GB，第 3 次 12GB
    time   = { 2.h * task.attempt }                 // 第 1 次 2h，第 2 次 4h，第 3 次 6h

    // 加上上限限制，防止无限增长
    cpus   = { Math.min(2 * task.attempt, params.max_cpus as int) }
    memory = { Math.min(4.GB * task.attempt, params.max_memory as nextflow.util.MemoryUnit) }
    time   = { Math.min(2.h * task.attempt, params.max_time as nextflow.util.Duration) }
}

process {
    // === 策略 1：retry（重试）===
    // 失败了自动重试，最常用
    errorStrategy = 'retry'
    maxRetries    = 3                               // 最多重试 3 次

    // === 策略 2：ignore（忽略）===
    // 失败了跳过这个样本，继续跑其他样本
    // 适合：某些样本质量差，不想因为一个坏样本卡住整个流程
    errorStrategy = 'ignore'

    // === 策略 3：finish（完成后停止）===
    // 出错后不提交新任务，但让已提交的任务跑完
    // 适合：发现系统问题，优雅停止
    errorStrategy = 'finish'

    // === 策略 4：terminate（立即终止）===
    // 出错立刻杀死所有任务
    // 适合：严重错误，必须马上停
    errorStrategy = 'terminate'

    // === 动态策略（推荐）===
    // 前几次重试，重试用完了就忽略
    errorStrategy = { task.attempt <= 3 ? 'retry' : 'ignore' }
    maxRetries    = 3
}

process {
    maxRetries = 3                                  // 单个任务最多重试 3 次
    maxErrors  = 5                                  // 整个流程最多允许 5 个错误
    // 超过 maxErrors 后，即使 errorStrategy='ignore'，流程也会停止
}

// 本地执行器（默认）
process.executor = 'local'

executor {
    cpus   = 4                                      // 本地最多用 4 核
    memory = '8.GB'                                 // 本地最多用 8GB 内存
}

// SLURM 执行器配置
process {
    executor       = 'slurm'                        // 使用 SLURM 调度
    queue          = 'normal'                       // SLURM 分区/队列名
    clusterOptions = '--account=mylab_account'      // 账户名（有些集群需要）
}

executor {
    queueSize        = 50                           // 最多同时 50 个作业排队
    pollInterval     = '1 min'                      // 每分钟检查作业状态
    queueStatInterval = '5 min'                     // 每 5 分钟查询队列状态
    submitRateLimit  = '10/1min'                    // 限速：每分钟最多提交 10 个
    perCpuMemAllocation = true                      // 有些集群要求 --mem-per-cpu
}

// AWS Batch 执行器配置
process {
    executor  = 'awsbatch'                          // AWS Batch 执行
    queue     = 'my-nextflow-queue'                 // Batch 队列名
    container = 'ubuntu:22.04'                      // 默认 Docker 容器
}

aws {
    region    = 'us-east-1'                         // AWS 区域
    accessKey = '...'                               // 通常用环境变量或 IAM Role
    secretKey = '...'                               // 不要硬编码在配置文件里！
    batch {
        cliPath = '/home/ec2-user/miniconda/bin/aws'  // AWS CLI 路径
        volumes = '/data'                           // 挂载的 EBS 卷
    }
}

// AWS Batch 必须用 S3 作为工作目录
workDir = 's3://my-bucket/nextflow-work'

// Kubernetes 执行器配置
process {
    executor  = 'k8s'                               // Kubernetes 执行
    container = 'ubuntu:22.04'                      // 每个 pod 的默认容器
}

k8s {
    namespace        = 'nextflow'                   // K8s 命名空间
    serviceAccount   = 'nextflow-sa'                // 服务账户
    storageClaimName = 'nextflow-pvc'               // 持久卷声明（共享存储）
    storageMountPath = '/workspace'                 // 挂载路径
}

# 生成 HTML 格式的执行报告
nextflow run main.nf -with-report report.html

# 报告内容包括：
# - 每个 process 的运行状态（成功/失败/跳过）
# - CPU 使用率、内存峰值、运行时间
# - 读写 I/O 统计
# - 整体流程概览

# 生成时间线报告（甘特图）
nextflow run main.nf -with-timeline timeline.html

# 直观展示：
# - 每个任务的开始和结束时间
# - 任务之间的并行关系
# - 哪个步骤是瓶颈

# 生成详细的资源追踪文件（TSV 格式）
nextflow run main.nf -with-trace trace.tsv

# 每行一个任务，记录：
# - task_id, name, status（任务 ID、名称、状态）
# - realtime, %cpu, rss, peak_rss（运行时间、CPU%、内存使用）
# - read_bytes, write_bytes（读写量）
# - submit, start, complete（提交、开始、完成时间）

# 生成有向无环图（流程依赖关系图）
nextflow run main.nf -with-dag flowchart.png

# 支持格式：png, svg, pdf, html
nextflow run main.nf -with-dag flowchart.svg

# 最佳实践：每次运行都带上所有报告参数
nextflow run main.nf \
    -with-report  results/report.html \            # 执行报告
    -with-timeline results/timeline.html \         # 时间线
    -with-trace   results/trace.tsv \              # 资源追踪
    -with-dag     results/dag.svg \                # DAG 图
    -profile docker                                # 运行 profile

# 第一步：注册 Seqera Platform（学术机构免费）
# 访问 https://seqera.io/ 注册账户

# 第二步：获取 Access Token
# 在 Seqera 网站 → Settings → Your Tokens → 创建新 Token

# 第三步：设置环境变量
export TOWER_ACCESS_TOKEN="你的token"              # 设置 Token

# 第四步：运行流程时加 -with-tower 参数
nextflow run main.nf -with-tower                   # 运行信息自动同步到 Seqera

# 也可以在 nextflow.config 里永久配置
tower {
    enabled     = true                              // 启用 Seqera Platform
    accessToken = "${TOWER_ACCESS_TOKEN}"            // Token（用环境变量，别硬编码）
}

ERROR ~ Error executing process > 'KRAKEN2 (sample_01)'
Process `KRAKEN2 (sample_01)` exceeded memory limit: 32 GB

// 方案 1：在 config 里给 KRAKEN2 加内存
process {
    withName: 'KRAKEN2' {
        memory = '64.GB'                            // 增加到 64GB
    }
}

// 方案 2：用动态资源 + 重试（推荐）
process {
    withName: 'KRAKEN2' {
        memory = { 32.GB * task.attempt }           // 第一次 32GB，重试 64GB
        errorStrategy = 'retry'
        maxRetries = 2
    }
}

ERROR ~ Error submitting process 'FASTP' to SLURM
sbatch: error: Batch job submission failed: Invalid account

# 先查看你能用的账户和队列
sacctmgr show assoc user=$USER format=Account,Partition  # 查看可用账户和分区
sinfo                                               # 查看集群队列信息

# 然后在 config 里写正确的信息
process {
    executor       = 'slurm'
    queue          = 'bio'                          # 改成你实际能用的队列
    clusterOptions = '--account=bio_group'          # 改成你实际的账户
}

ERROR ~ Error executing process > 'FASTP'
Unable to pull image 'quay.io/biocontainers/fastp:0.99.0'

# 到 BioContainers 网站搜索正确的镜像名和版本
# https://biocontainers.pro/

# 或者用命令搜索
docker search fastp                                 # 搜索 Docker Hub
singularity search fastp                            # 搜索 Singularity

# 在模块文件里修正镜像地址
container "quay.io/biocontainers/fastp:0.24.0--heae7151_0"  # 用正确的版本

# 跑了 -profile slurm 但还是在本地执行
nextflow run main.nf -profile slurm

# 检查拼写
# 注意：是 -profile（一个横杠），不是 --profile（两个横杠）
nextflow run main.nf -profile slurm                 # 正确
nextflow run main.nf --profile slurm                # 错误！这会变成 params.profile

# 查看 config 里定义了哪些 profile
grep -A2 "profiles" nextflow.config                 # 确认 profile 名字

# 基本运行
nextflow run main.nf                               # 默认配置运行

# 指定 profile
nextflow run main.nf -profile docker               # Docker 模式
nextflow run main.nf -profile slurm,singularity    # SLURM + Singularity

# 覆盖参数
nextflow run main.nf --reads "data/*.fq.gz" --outdir results

# 恢复中断的运行
nextflow run main.nf -resume                       # 从上次中断处继续

# 带所有报告
nextflow run main.nf -with-report -with-timeline -with-trace -with-dag dag.svg

# 后台运行
nextflow run main.nf -bg > nf.log 2>&1             # 后台执行，日志写文件

# 查看运行历史
nextflow log                                        # 列出历史运行记录

# 清理工作目录
nextflow clean -f                                   # 删除 work/ 目录释放空间