Dagster Pipes with existing Databricks jobs

Preview feature

This feature is considered in a preview stage, and is under active development, and not considered ready for production use. You may encounter feature gaps, and the APIs may change. For more information, see the API lifecycle stages documentation.

If your work is already defined as a Databricks job (workflow), you can trigger it by ID with Dagster Pipes instead of rebuilding its tasks in Dagster. The job keeps its own tasks and compute; Dagster passes it a Pipes context and receives logs and metadata back.

Both Pipes clients support this with run_job:

Client	Pipes storage	Use with
PipesDatabricksServerlessClient	Unity Catalog Volume	Serverless jobs, and classic jobs on Unity Catalog-enabled compute
PipesDatabricksClient	Databricks File System (DBFS)	Classic jobs in workspaces where DBFS is available

To submit one-time runs from task definitions in Dagster instead, see serverless compute or DBFS, classic clusters.

Prerequisites

• The dagster-databricks library installed in your Dagster project, and the DATABRICKS_HOST and DATABRICKS_TOKEN environment variables set. See serverless compute.
• dagster-pipes installed in the job's environment.
• An existing Databricks job, and its job ID.

Step 1: Add the Pipes parameters to your job

Dagster passes the Pipes context to the job as two job parameters: DAGSTER_PIPES_CONTEXT and DAGSTER_PIPES_MESSAGES. In the job's settings, declare both with an empty default value:

"parameters": [
  {"name": "DAGSTER_PIPES_CONTEXT", "default": ""},
  {"name": "DAGSTER_PIPES_MESSAGES", "default": ""}
]

How a task receives these parameters depends on its type:

• Notebook tasks receive job parameters automatically as widgets. No further setup is needed.

• Python script tasks only receive what's listed in their task parameters. Forward both Pipes parameters as arguments:

  "spark_python_task": {
    "python_file": "/Workspace/path/to/script.py",
    "parameters": [
      "--dagster-pipes-context", "{{job.parameters.DAGSTER_PIPES_CONTEXT}}",
      "--dagster-pipes-messages", "{{job.parameters.DAGSTER_PIPES_MESSAGES}}"
    ]
  }

note

All tasks in the run receive the same Pipes parameters, so only one task per run should report back to Dagster with Pipes.

Step 2: Add Pipes to the task code

Wrap the task's logic with open_dagster_pipes. The context loader and message writer must match your Dagster client: the Unity Catalog Volumes loaders for PipesDatabricksServerlessClient, or PipesDbfsContextLoader and PipesDbfsMessageWriter for PipesDatabricksClient.

Notebook task

Read the Pipes parameters from the notebook widgets with PipesDatabricksNotebookWidgetsParamsLoader:

Databricks notebook

from dagster_pipes import (
    PipesDatabricksNotebookWidgetsParamsLoader,
    PipesUnityCatalogVolumesContextLoader,
    PipesUnityCatalogVolumesMessageWriter,
    open_dagster_pipes,
)

with open_dagster_pipes(
    context_loader=PipesUnityCatalogVolumesContextLoader(),
    message_writer=PipesUnityCatalogVolumesMessageWriter(),
    params_loader=PipesDatabricksNotebookWidgetsParamsLoader(dbutils.widgets),
) as pipes:
    # Access parameters passed from Dagster
    some_parameter = pipes.get_extra("some_parameter")
    pipes.log.info(f"Running with some_parameter={some_parameter}")

    # ... your notebook computation ...

    # Report structured metadata back to Dagster
    pipes.report_asset_materialization(
        metadata={
            "row_count": {"raw_value": 99_000, "type": "int"},
            "output_table": {"raw_value": "catalog.schema.output", "type": "text"},
        }
    )

Python script task

Read the Pipes parameters from the command-line arguments with PipesCliArgsParamsLoader:

Databricks Python script

from dagster_pipes import (
    PipesCliArgsParamsLoader,
    PipesUnityCatalogVolumesContextLoader,
    PipesUnityCatalogVolumesMessageWriter,
    open_dagster_pipes,
)

Step 3: Trigger the job from Dagster

Call run_job with the job ID:

src/<project-name>/defs/databricks-assets.py

import os

from dagster_databricks.pipes import PipesDatabricksServerlessClient
from databricks.sdk import WorkspaceClient

import dagster as dg


@dg.asset
def feature_engineering(
    context: dg.AssetExecutionContext,
    pipes_databricks: PipesDatabricksServerlessClient,
) -> dg.MaterializeResult:
    return pipes_databricks.run_job(
        context=context,
        job_id=int(os.environ["DATABRICKS_JOB_ID"]),
        # Optional: job parameters for this run, alongside the Pipes params
        job_parameters={"target_schema": "analytics"},
        extras={"some_parameter": 42},
    ).get_materialize_result()


@dg.definitions
def resources():
    pipes_databricks_resource = PipesDatabricksServerlessClient(
        client=WorkspaceClient(
            host=os.environ["DATABRICKS_HOST"],
            token=os.environ["DATABRICKS_TOKEN"],
        ),
        # A Unity Catalog Volume used as the Pipes message transport
        volume_path="/Volumes/catalog/schema/dagster_pipes",
    )
    return dg.Definitions(
        assets=[feature_engineering],
        resources={"pipes_databricks": pipes_databricks_resource},
    )

run_job waits for the job run to finish, and the asset fails if the run fails. If the Dagster run is canceled, the Databricks run is canceled too. Your own job_parameters are passed alongside the Pipes parameters, and run_now_args passes other options to WorkspaceClient.jobs.run_now, such as only to run a subset of tasks or performance_target.

Triggering jobs without Pipes

If you can't add Pipes to the job's code, you can still trigger the job from Dagster and wait for it to finish, without streaming logs or metadata back. See Triggering existing jobs without Pipes, or use the Databricks workspace component to model your jobs as assets.