nextflow-workflow-engine

Nextflow — スケーラブルな科学ワークフローエンジン

概要

Nextflow は、プロセス（コンテナ化された実行単位）がチャネル（非同期キュー）を介してデータを受け取り、出力するデータフロープログラミングモデルを実装しています。この設計により、暗黙的な並列化が可能になります。入力チャネルにデータが届くと、手動で依存関係を管理しなくてもプロセスが実行されます。Nextflow は、設定プロファイルを1つ切り替えるだけで、ローカルマシン、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 との統合を優先する Python ネイティブのルールベースワークフローには、代わりに Snakemake を使用する
• CWL/WDL 規格への準拠が必要な臨床ゲノミクスパイプラインには、代わりに WDL/Cromwell を使用する

前提条件

• ソフトウェア：Java 11+、Nextflow（自己完結型ランチャー）
• コンテナ：プロセスを分離するための Docker または Singularity（推奨）
• 任意：コミュニティパイプラインを管理するための nf-core ツール

インストール前に確認：現在の環境（例：pixi / conda 環境内）ですでにツールを利用できる場合があります。まず command -v nextflow を実行し、パスが返された場合は、以下のインストールコマンドを実行しないでください。pixi プロジェクト内で実行する場合は、nextflow をそのまま実行するのではなく、pixi run 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：プロセス — コンテナ化されたタスク単位

入力、出力、シェル／スクリプトのディレクティブを指定してプロセスを定義します。

// 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 et al. (2017)「Nextflow enables reproducible computational workflows」— Nature Biotechnology 35:316-319。DOI:10.1038/nbt.3820
• nf-core — 自動テストとコンテナーを備えた、コミュニティが厳選した Nextflow パイプライン