Summary

This plugin integrates Nextflow with lakeFS, allowing you to use lakefs:// URIs as inputs and outputs in your pipelines. It brings the power of data versioning, reproducibility, and atomic operations to your Nextflow workflows. The plugin treats lakeFS repositories and branches as a native file system, enabling seamless data management without requiring manual pre-loading or post-processing steps.

Key Features:

• Native URI Support: Use lakefs://<repo>/<branch>/<path> URIs directly in your Nextflow scripts and nextflow.config
• Read Inputs: Stage input files from a lakeFS repository for use in your processes
• Publish Outputs: Publish output files from your processes directly to a lakeFS repository branch
• Transparent Authentication: Securely configure your lakeFS credentials
• Multiple Transfer Modes: Support for both signed_url and physical_path modes
• Auto-commit Support: Automatically commit changes on successful workflow completion

Get Started

Installation

Add the following to your nextflow.config file. Replace <version> with the desired plugin version (e.g., 0.1.0):

plugins {
    id 'nf-lakefs@<version>'
}

Configuration

Configure the plugin with your lakeFS server details and credentials in your nextflow.config:

lakefs {
    apiUrl    = 'https://your-lakefs-server.example.com/api/v1'
    accessKey = 'YOUR_LAKEFS_ACCESS_KEY'
    secretKey = 'YOUR_LAKEFS_SECRET_KEY'
    transferMode = 'signed_url'
    autoCreateBranch = false
    autoCreateBranchSource = 'main'
}

Configuration Options:

Option	Required	Default	Description
apiUrl	Yes	-	lakeFS API endpoint URL (must end with /api/v1)
accessKey	Yes	-	lakeFS access key
secretKey	Yes	-	lakeFS secret key
transferMode	No	signed_url	Transfer mode: signed_url or physical_path
autoCreateBranch	No	false	Automatically create branch if it doesn't exist when writing
autoCreateBranchSource	No	main	Source branch to create new branches from
readTimeout	No	60s	HTTP read timeout for lakeFS API calls
connectTimeout	No	30s	HTTP connect timeout for lakeFS API calls
allowedSchemaBucketsForLinking	No	[]	Whitelist of cloud storage scheme+bucket prefixes (e.g. s3://bucket, gs://bucket) allowed as link sources when the source bucket differs from the repository's backing bucket
autoCommit	No	false	Automatically commit the branch backing the workflow output directory once the workflow completes successfully

For S3-backed repositories with physical_path mode, also configure AWS credentials:

aws {
    accessKey = '<AWS_ACCESS_KEY_ID>'
    secretKey = '<AWS_SECRET_ACCESS_KEY>'
    region    = 'eu-central-1'
}

Auto-Create Branch Example

When autoCreateBranch is enabled, new branches are automatically created if they don't exist:

lakefs {
    apiUrl    = 'https://your-lakefs-server.example.com/api/v1'
    accessKey = 'YOUR_LAKEFS_ACCESS_KEY'
    secretKey = 'YOUR_LAKEFS_SECRET_KEY'
    autoCreateBranch = true
    autoCreateBranchSource = 'main'
}

Writing to lakefs://my-repo/new-branch/file.txt will automatically create new-branch from main if it doesn't already exist.

allowedSchemaBucketsForLinking Configuration

When using the physical_path transfer mode with different backing storage, whitelist compatible cloud storage backends:

lakefs {
    allowedSchemaBucketsForLinking = [
            'gs://bucket1',
            'gs://bucket2'
    ]
}

Or for S3-backed repositories:

lakefs {
    allowedSchemaBucketsForLinking = [
            's3://bucket1',
            's3://bucket2'
    ]
}

Examples

Basic Pipeline with lakeFS Input and Output

Create a main.nf pipeline that reads a file from a lakeFS repository, processes it, and publishes the result:

params.input = 'lakefs://my-repo/main/path/to/**/config.yaml'
params.output_dir = 'lakefs://my-repo/output/path/to/'

process EXAMPLE_PROCESS {

    input:
    path my_file

    output:
    path 'result.txt'

    script:
    """
    echo "Processing ${my_file}..."
    cat "${my_file}" | wc -l > result.txt
    """
}

workflow {
    Channel.fromPath(params.input, type: 'file')
            | view
            | EXAMPLE_PROCESS
}

Run the pipeline:

nextflow run main.nf

Automatic Commit on Workflow Completion

Enable automatic commits by setting autoCommit = true in your configuration:

lakefs {
    apiUrl    = 'https://your-lakefs-server.example.com/api/v1'
    accessKey = 'YOUR_LAKEFS_ACCESS_KEY'
    secretKey = 'YOUR_LAKEFS_SECRET_KEY'
    autoCommit = true
}

outputDir = 'lakefs://my-repo/output-branch/path/to/'

Run the pipeline with a commit message:

nextflow run main.nf --lakefs_commit_message "my commit message"

Manual Commit Control

For finer-grained control, use the lakefsCommit() function to commit specific branches:

workflow.onComplete {
    if (workflow.success) {
        lakefsCommit(params.output_dir, params.lakefs_commit_message)
    }
}

The path parameter can be any lakefs://<repo>/<branch>/... URI; only the repository and branch components are used. The function returns the created commit ID.

License

This Nextflow plugin is distributed under the Apache 2.0 License.