Skip to main content
This guide will walk you through creating a simple Metaflow flow and deploying it to Dagster.

​
Prerequisites

Make sure you have metaflow-dagster and dagster installed: 
pip install metaflow-dagster dagster dagster-webserver

​
Step-by-Step Tutorial

1

Create a Metaflow Flow

Create a new file called my_flow.py with a simple linear flow:
my_flow.py
from metaflow import FlowSpec, step

class MyFlow(FlowSpec):
    """A simple flow to demonstrate Dagster integration."""

    @step
    def start(self):
        """Start step: initialize data."""
        self.message = "Hello from Metaflow!"
        print(f"Starting flow with message: {self.message}")
        self.next(self.process)

    @step
    def process(self):
        """Process step: transform the data."""
        self.result = self.message.upper()
        print(f"Processed result: {self.result}")
        self.next(self.end)

    @step
    def end(self):
        """End step: finalize the flow."""
        print(f"Flow completed with result: {self.result}")

if __name__ == "__main__":
    MyFlow()
You can test your flow locally first using standard Metaflow commands:
python my_flow.py run
2

Generate Dagster Definitions

Use the dagster create command to compile your flow into a Dagster definitions file:
python my_flow.py dagster create my_flow_dagster.py
You should see output like:
Compiling MyFlow to Dagster job MyFlow...
Dagster job MyFlow for flow MyFlow written to my_flow_dagster.py.
Load it in Dagster with:
    dagster dev -f my_flow_dagster.py
This command generates a self-contained Python file (my_flow_dagster.py) that contains:
  • Dagster @op definitions for each step
  • A @job that wires the ops together
  • A Definitions object that Dagster can load
3

Launch Dagster Dev Server

Start the Dagster development server to view and run your flow:
dagster dev -f my_flow_dagster.py
You should see output indicating the server is running:
Serving dagster-webserver on http://127.0.0.1:3000 in process 12345
Open your browser to http://localhost:3000 to access the Dagster UI.
4

View Your Flow in the Dagster UI

In the Dagster UI:
  1. Click on “Jobs” in the left sidebar
  2. Select your job (MyFlow)
  3. You’ll see a visual representation of your flow graph showing the three steps: start → process → end
  4. Click “Materialize” (or “Launch run”) to execute the flow
The Dagster UI provides:
  • Job Graph: Visual representation of your flow’s DAG
  • Launchpad: Configuration interface for parameters
  • Run Timeline: Gantt chart showing execution timing
  • Logs: Real-time streaming logs from each step
  • Metadata: Artifact keys and retrieval snippets
5

Execute Your Flow

Click the “Materialize” button in the Dagster UI to run your flow. You’ll see:
  1. Real-time progress: Each step turns green as it completes
  2. Execution logs: Click on any step to see its output
  3. Timing information: See how long each step took
  4. Artifact metadata: Each step shows the Metaflow artifacts it produced
After the run completes, you can click on individual steps to see:
  • The exact metaflow step CLI command that was executed
  • Your flow’s print() output
  • Artifact keys and a ready-to-copy retrieval snippet

​
Working with Parameters

Metaflow Parameter definitions are automatically converted to Dagster configuration. Here’s an example:
parametrized_flow.py
from metaflow import FlowSpec, Parameter, step

class ParametrizedFlow(FlowSpec):
    """A flow that uses Metaflow Parameters."""

    greeting = Parameter("greeting", default="Hello", help="Greeting prefix")
    count = Parameter("count", default=3, type=int, help="Number of repetitions")

    @step
    def start(self):
        self.messages = [f"{self.greeting} #{i}" for i in range(self.count)]
        print(f"Generated {len(self.messages)} messages")
        self.next(self.end)

    @step
    def end(self):
        for msg in self.messages:
            print(msg)

if __name__ == "__main__":
    ParametrizedFlow()
Compile and run: 
python parametrized_flow.py dagster create param_flow_dagster.py
dagster dev -f param_flow_dagster.py
In the Dagster UI’s Launchpad, you’ll see a configuration form with your parameters. You can override the defaults before launching the run.
Important: All Metaflow parameters must have default values when deploying to Dagster. This ensures the flow can be executed without manual intervention.

​
Running from the CLI

You can also execute your Dagster job directly from the command line: 
# Execute with default parameters
python -m dagster job execute -f my_flow_dagster.py -j MyFlow

# Execute with custom parameters (requires a config file)
cat > config.yaml << EOF
ops:
  op_start:
    config:
      greeting: Hi
      count: 5
EOF

python -m dagster job execute -f param_flow_dagster.py -j ParametrizedFlow -c config.yaml

​
Accessing Artifacts

After a run completes, you can access Metaflow artifacts from the Dagster UI or programmatically:

​
From the Dagster UI

Click on any completed step to see its metadata panel. You’ll find:
  • Artifact keys: The names of all artifacts created by this step
  • Retrieval snippet: Ready-to-copy Python code to load the artifacts

​
Programmatically

Use the standard Metaflow client to access artifacts: 
from metaflow import Task

# Replace with your actual run ID (visible in Dagster UI)
task = Task('MyFlow/dagster-abc123/process/1')
result = task['result'].data
print(result)  # "HELLO FROM METAFLOW!"

​
Advanced Features

​
Step Decorators

Inject Metaflow step decorators at deploy time without modifying your flow source: 
# Run every step in a sandbox
python my_flow.py dagster create my_flow_dagster.py --with=sandbox

# Add resource hints
python my_flow.py dagster create my_flow_dagster.py \
  --with='resources:cpu=4,memory=8000'

​
Workflow Timeout

Set a maximum wall-clock time for the entire job run: 
python my_flow.py dagster create my_flow_dagster.py --workflow-timeout 3600

​
Custom Job Name

Override the default job name: 
python my_flow.py dagster create my_flow_dagster.py --name nightly_pipeline

​
Tags

Add Metaflow tags that are forwarded to every step subprocess: 
python my_flow.py dagster create my_flow_dagster.py \
  --tag env:prod \
  --tag version:2

​
What You Learned

Flow Creation

How to write a Metaflow flow that works with Dagster

Compilation

How to generate Dagster definitions from a Metaflow flow

Execution

How to run your flow in the Dagster UI and CLI

Parameters

How to use Metaflow Parameters with Dagster configuration

​
Next Steps

Now that you’ve created your first flow, explore more advanced features:

Graph Patterns

Learn about branching, conditional, and foreach flows

Scheduling

Set up automated flow execution with Dagster schedules

Configuration

Configure metadata services, datastores, and production settings

Examples

Explore complete working examples