Nextflow

The Nextflow plugin runs Nextflow pipelines on Flyte. The Nextflow head runs inside a Flyte task, and the nf-flyte Nextflow plugin runs every Nextflow task as a child action of that task. A pipeline therefore shows up in Flyte as one run, with one action per Nextflow task:

• Each process runs in its own container, with its cpus, memory, disk, accelerator, and time directives. The image only needs bash: Flyte stages the task’s files.
• Nextflow keeps its own retries (errorStrategy) and -resume. Flyte caches task results across runs.
• Aborting the Flyte run aborts the pipeline and its running tasks.

Use this plugin to run existing Nextflow pipelines, such as those from nf-core, without rewriting them, and to chain them with Python tasks in the same Flyte run.

Installation

pip install flyteplugins-nextflow

Quick start

nextflow_image() builds the image for the task that runs the Nextflow head, with flyteplugins-nextflow installed. Call run_nextflow from an async task:

import flyte
from flyte.io import Dir
from flyteplugins.nextflow import nextflow_image, run_nextflow

env = flyte.TaskEnvironment(
    name="nextflow",
    image=nextflow_image(),
    resources=flyte.Resources(cpu=2, memory="4Gi"),
)


@env.task
async def nf_core_demo(revision: str = "1.2.0") -> Dir:
    # The `test` profile brings its own small samplesheet and resource limits
    return await run_nextflow("nf-core/demo", revision=revision, profile="test", outdir="results")

flyte run nf_core_demo.py nf_core_demo

This runs the nf-core/demo pipeline (FastQC, seqtk trim, and MultiQC). Each process runs as a child action of nf_core_demo, in its own biocontainers image, and the results directory comes back as a Dir.

The resources on the task environment apply only to the Nextflow head. Pipeline processes get their resources from their own directives.

run_nextflow arguments

run_nextflow must be called from inside a Flyte task.

For example, to pass pipeline parameters and add a config override:

await run_nextflow(
    "nf-core/rnaseq",
    params={"input": "s3://my-bucket/samplesheet.csv", "genome": "GRCh38"},
    config="process.errorStrategy = 'retry'\nprocess.maxRetries = 2",
    outdir="results",
)

Reports

With report=True, Nextflow’s execution report, timeline, and DAG appear as tabs in the task’s report in the Flyte UI. Declare the task with report=True as well:

@env.task(report=True)
async def rnaseq() -> Dir:
    return await run_nextflow("nf-core/rnaseq", profile="test", outdir="results", report=True)

The reports are published even when the pipeline fails.

Resume

Nextflow’s resume state lives in the work directory, not in the task pod:

• the task cache (the cloud cache on object stores; .nextflow-cache in the work directory locally)
• a session ID derived from the work directory

This has two effects:

• Retries resume automatically. The default work directory is shared by all attempts of a task, so a retried task passes -resume and picks up where the failed attempt stopped.

• Resuming across runs needs a fixed work_dir. Pass the same work_dir to each run, and resume=True from the second run on:

  await run_nextflow("nf-core/rnaseq", work_dir="s3://my-bucket/rnaseq-work", resume=True)

Errors

If Nextflow exits with a non-zero status, run_nextflow raises a NextflowError. Its message contains the end of the Nextflow output and the end of .nextflow.log, and its exit_code attribute holds Nextflow’s exit status.

Local runs

When the task runs locally (flyte run --local), Nextflow uses its own executors instead of nf-flyte: local by default, or whatever profile or config selects, such as docker. This needs nextflow on the PATH.

Image

nextflow_image() contains a JRE, Nextflow, and the nf-flyte plugin. Pipeline processes don’t run in this image; each one runs in its own container.

To use a locally built nf-flyte plugin, pass its zip:

from pathlib import Path

image = nextflow_image(
    nf_flyte=Path("~/nf-flyte/build/distributions/nf-flyte-0.1.0.zip").expanduser(),
)

Requirements

• flyte 2.12.0 or later, and flyteplugins-nextflow 2.12.0 or later.
• On a cluster, the work directory must be on an object store Flyte can read (S3, GCS, or Azure). The default, under the task’s raw data prefix, always is.
• Every process container needs bash.