06 - Dagster Pipes, dbt, And External Systems

8 min read

Dagster Pipes, dbt, And External Systems

Dagster becomes most valuable in mixed-compute systems when it refuses to impersonate the runtime that should actually execute the work. The orchestrator should know what data state was produced, what metadata came back, and which downstream assets now depend on it. That does not mean every warehouse transform, Spark job, or review export needs to run inside the Dagster process itself.

Dagster | Pipes

Dagster Pipes exists for workloads that should stay in another process or execution environment while still reporting structured metadata and logs back to Dagster. The goal is not to absorb remote compute into the control plane. It is to keep the runtime boundary explicit and observable.

Dagster | external runtimes | observe remote compute without absorbing it

If the compute belongs in another process, container, cluster, or managed platform, Dagster should not pretend that boundary does not exist. It should coordinate the work, receive results and metadata, and keep orchestration state legible without taking ownership of the runtime’s internals.

Pipes | launch a subprocess through Dagster Pipes

Use Pipes when the compute environment is intentionally separate from the Dagster process. The trigger is work that needs its own dependency set, system image, or runtime isolation. The asset runs in Dagster, but the real compute happens in another process launched through a Pipes client. Its purpose is to let the external code remain external while still reporting structured metadata and logs back to Dagster.

Define an asset that delegates work to a child process through PipesSubprocessClient.

from pathlib import Path
import sys
 
import dagster as dg
from dagster._core.pipes.subprocess import PipesSubprocessClient
 
CHILD = Path("pipes_child.py")
 
@dg.asset
def remote_table(
    context: dg.AssetExecutionContext,
    pipes_client: PipesSubprocessClient,
):
    yield from pipes_client.run(
        context=context,
        command=[sys.executable, str(CHILD)],
    ).get_results()

This is the right model for a future index-composition or benchmark-build workload that might run in Spark, Databricks, or another specialized runtime while Dagster still needs the resulting asset state, metadata, and lineage. The local dagflow repository does not currently use Pipes, which is itself instructive: Pipes should appear only when the compute boundary is genuinely external, not as a default integration reflex.

Dagster | dbt integration

Dagster’s dbt integration is useful because it understands dbt work as structured topology: models, sources, tests, and checks with meaningful lineage and selective recovery. That is much stronger than treating dbt build as one opaque shell command with no internal graph.

Dagster | dbt resources and translators | structured asset topology

Once dbt is part of the platform, the important question is not only whether the command runs. It is whether Dagster can reason about the resulting nodes as first-class states in the asset graph, with useful keys, groups, tags, and checks.

DbtCliResource | construct the shared dbt resource

Use DbtCliResource when the project needs Dagster to orchestrate a real dbt project with an explicit executable, manifest state, and profile location. The trigger is a warehouse transformation surface that should become part of the platform’s controlled runtime. The resource is built in shared dependency configuration, not ad hoc inside one asset body. Its purpose is to publish the dbt boundary once and reuse it consistently across dbt-backed assets.

Build the shared dbt resource in dagflow alongside the control-plane resource.

def build_resources() -> dict[str, ConfigurableResource | DbtCliResource]:
    settings = get_settings()
    dbt_executable = Path(sys.executable).with_name("dbt")
    dbt_project = get_dbt_project()
    return {
        "control_plane": ControlPlaneResource(
            direct_database_url=settings.direct_database_url,
            export_root_dir=settings.export_root_dir,
            landing_root_dir=str(settings.resolved_landing_root_dir),
            edgar_identity=settings.edgar_identity,
        ),
        "dbt": DbtCliResource(
            project_dir=dbt_project,
            dbt_executable=str(dbt_executable),
        ),
    }

The important point is not that DbtCliResource exists. It is that the dbt project becomes part of the same explicit runtime contract as the rest of the Dagster resources. If the executable path, profiles directory, or project manifest changes, that change happens in one resource boundary instead of being rediscovered by each asset separately.

DagsterDbtTranslator | surface dbt nodes and tests as assets and checks

Use a translator when the team needs dbt nodes to land in Dagster with meaningful asset keys, groups, tags, and checks. The trigger is a project where raw dbt defaults do not communicate enough operational meaning about domain, layer, or ownership. The translator runs at definition time and shapes how Dagster sees the dbt project. Its purpose is to make dbt topology legible to Dagster operators rather than leaving it as a flat imported graph.

Translate dbt nodes into domain-prefixed asset keys and enable dbt tests as Dagster checks.

class DagflowDbtTranslator(DagsterDbtTranslator):
    def __init__(self) -> None:
        super().__init__(
            settings=DagsterDbtTranslatorSettings(
                enable_asset_checks=True,
                enable_source_tests_as_checks=True,
            )
        )
 
    def get_asset_key(self, dbt_resource_props: Mapping[str, Any]) -> dg.AssetKey:
        if dbt_resource_props.get("resource_type") == "source":
            return super().get_asset_key(dbt_resource_props)
        return dbt_asset_key(
            pipeline_for_resource(dbt_resource_props), str(dbt_resource_props["name"])
        )

This is where dagflow makes dbt assets operationally meaningful. dim_security does not appear as a detached dbt node. It appears inside a pipeline-prefixed asset namespace such as security_master__dim_security, alongside checks and metadata that tell the control plane where in the warehouse layer the node belongs.

Dagster | dbt source mapping | connect dbt to upstream Dagster state

Dagster’s dbt integration also depends on honest source mapping. When dbt sources refer to upstream Dagster assets through meta.dagster.asset_key, the graph can show that warehouse sources and review tables are the same operational states Dagster already knows elsewhere.

meta.dagster.asset_key | map dbt sources to upstream asset keys

Use source mapping when dbt models depend on data states that Dagster already models elsewhere, such as raw landing assets or governed review tables. The trigger is a need for lineage that crosses tool boundaries honestly. The mapping lives in dbt project metadata, not in a side spreadsheet or mental model. Its purpose is to let Dagster understand that a dbt source is the same operational state as an upstream Dagster asset.

Map raw tables and review tables to the Dagster assets that own those states.

version: 2
 
sources:
  - name: raw
    schema: raw
    tables:
      - name: sec_company_tickers
        meta:
          dagster:
            asset_key: ["security_master__sec_company_tickers_raw"]
 
  - name: review
    schema: review
    tables:
      - name: security_master_daily
        meta:
          dagster:
            asset_key: ["security_master__review_snapshot"]

That mapping is what lets dagflow express an export model as depending on the reviewed state, not only on the most recent warehouse transform. It is also what keeps the asset graph honest across the full governed lifecycle: raw landing, curated transforms, review snapshot, export preview, and final CSV delivery.

Dagster | governed external handoffs

Mixed-compute platforms stay understandable only when compute semantics and orchestration state remain in the right places. Dagster should decide what to run, record what happened, and expose lineage and checks. The external system should remain the owner of its own compute semantics.

Dagster | delivery boundaries | keep compute semantics and orchestration state separate

The best external-system integrations do not hide the handoff between systems. They make it explicit enough that blast radius, reruns, and state transitions remain explainable across review, transformation, and delivery boundaries.

Asset jobs | build export work from reviewed state

Use this pattern when the pipeline includes a governed review boundary between transformation and delivery. The trigger is any workflow where a human-approved or policy-approved state is materially different from the latest machine-computed state. The export asset is still orchestrated by Dagster, but it should depend on the reviewed source of truth. Its purpose is to ensure that delivery reflects the approved dataset, not merely the most recent transform run.

Select dbt export assets and CSV export assets from the reviewed state boundary in one asset job.

security_master_export_job = define_asset_job(
    name="security_master_export_job",
    executor_def=in_process_executor,
    selection=AssetSelection.assets(security_master_csv_export)
    | build_dbt_asset_selection([security_master_export_assets]),
)

The approved review state is the real delivery contract

In dagflow, the export models read from review tables that are mapped back to Dagster review-snapshot assets. That means the exported security master or shareholder holdings file is not “whatever dbt most recently computed.” It is the reviewer-approved state that survived a governed handoff. That is exactly the kind of cross-system boundary Dagster should make visible rather than flatten into one shell step.

Dagster | references

This section collects the official Dagster documentation links most relevant to Pipes, dbt integration, and external-system boundaries.

Graph View

Backlinks