Step 4 — Compose your first pipeline

OrcheStack doesn't ship a one-size-fits-all pipeline. It ships the building blocks. This page shows three concrete paths through those blocks — pick the one closest to your stack, copy the snippet, adapt to your data, run.

Choose your stack

The platform's transformation layer is opinionated — dbt runs from Airflow with per-model task granularity via astronomer-cosmos. Every other choice is yours: which ingestion tool, which BI tool, which scheduling cadence, even whether to have BI at all.

Three patterns cover most operator stacks. Each pattern is a complete Airflow DAG you can paste into your own dags/ folder, edit, and run.

Path	Ingest	Transform	BI / consumption	Read this if you…
A	Airbyte	dbt + Cosmos	Metabase	… want the classic ELT stack out of the box
B	Custom Python loader	dbt + Cosmos	External Tableau (or any BI tool outside OrcheStack)	… already have a BI tool and want to bring your own ingestion logic
C	dlt	dbt + Cosmos	None — engineers query the warehouse directly	… are a data team without analyst-facing BI yet

You can compose your own path. Mix the ingestion pattern from B with the BI pattern from A. Or run two ingestion sources in the same DAG. The patterns are independent — they share only dbt+Cosmos as the middle stage.

Path A: Airbyte → dbt → Metabase

The canonical modern-data-stack pipeline. Trigger an Airbyte connection, wait for the sync to finish, run dbt models (one Airflow task per model), refresh a Metabase dashboard so stakeholders see fresh data.

Prerequisites: Airbyte deployed (cold tier), at least one Airbyte connection configured. Metabase deployed (hot tier) with a dashboard. dbt project on disk (see Set up your dbt project).

from datetime import datetime
from pathlib import Path

from airflow import DAG
from airflow.providers.http.operators.http import HttpOperator
from airflow.providers.http.sensors.http import HttpSensor

from cosmos import DbtDag, ProjectConfig, ProfileConfig, ExecutionConfig
from cosmos.profiles import PostgresUserPasswordProfileMapping

DBT_PROJECT_PATH = Path("/opt/airflow/dbt-project")

# Cosmos profile config. Re-used across DAGs. The Airflow connection
# `orchestack_warehouse` is auto-created on first Airflow start by
# the orchestrator's post-start hook — no manual setup needed.
profile_config = ProfileConfig(
    profile_name="orchestack_warehouse",
    target_name="prod",
    profile_mapping=PostgresUserPasswordProfileMapping(
        conn_id="orchestack_warehouse",
        profile_args={"schema": "public"},
    ),
)

with DAG(
    dag_id="path_a_airbyte_dbt_metabase",
    start_date=datetime(2026, 1, 1),
    schedule="@daily",
    catchup=False,
    tags=["composition-pattern"],
) as dag:

    trigger_airbyte = HttpOperator(
        task_id="trigger_airbyte_sync",
        http_conn_id="airbyte",                # connection set in the Airflow UI
        endpoint="api/v1/connections/sync",
        method="POST",
        data='{"connectionId": "{{ var.value.airbyte_connection_id }}"}',
    )

    wait_for_airbyte = HttpSensor(
        task_id="wait_for_airbyte",
        http_conn_id="airbyte",
        endpoint="api/v1/jobs/{{ ti.xcom_pull(task_ids='trigger_airbyte_sync')['job']['id'] }}",
        response_check=lambda r: r.json()["job"]["status"] == "succeeded",
        poke_interval=30,
        timeout=60 * 60,
    )

    # Cosmos generates one task per dbt model + per dbt test.
    # If your project has 12 models and 30 tests, this expands to 42
    # Airflow tasks in dependency order. Failures attach to the
    # specific failing model, not to a single opaque dbt-run task.
    dbt_models = DbtDag(
        dag_id_prefix="dbt",
        project_config=ProjectConfig(DBT_PROJECT_PATH),
        profile_config=profile_config,
        execution_config=ExecutionConfig(dbt_executable_path="/home/airflow/.local/bin/dbt"),
    )

    refresh_metabase = HttpOperator(
        task_id="refresh_metabase",
        http_conn_id="metabase",
        endpoint="api/dashboard/{{ var.value.metabase_dashboard_id }}/refresh",
        method="POST",
    )

    trigger_airbyte >> wait_for_airbyte >> dbt_models >> refresh_metabase

Setup before triggering: in the Airflow UI, add two HTTP connections (airbyte pointing at http://orchestack-airbyte:8000, metabase pointing at http://orchestack-metabase:3000) and two Airflow Variables (airbyte_connection_id, metabase_dashboard_id).

Path B: Python ingest → dbt → Tableau (external)

You already have Tableau (or any BI tool outside OrcheStack — Power BI, Looker, Sigma). Your ingestion is custom Python that calls some API and lands rows in PostgreSQL. dbt transforms them. Tableau's refresh API picks up the rest.

The pattern uses Airflow's PythonVirtualenvOperator — it creates a per-task virtualenv with whatever Python libraries you list, isolated from Airflow's main environment. First run installs deps (~30-60s); subsequent runs reuse the cached venv.

from datetime import datetime
from pathlib import Path

from airflow import DAG
from airflow.operators.python import PythonVirtualenvOperator
from airflow.providers.http.operators.http import HttpOperator

from cosmos import DbtDag, ProjectConfig, ProfileConfig, ExecutionConfig
from cosmos.profiles import PostgresUserPasswordProfileMapping


def ingest_to_warehouse():
    # Runs in an ephemeral venv with `requests` + `psycopg2-binary`
    # available. Replace this with your actual loading logic.
    import requests
    import psycopg2

    response = requests.get("https://api.example.com/data", timeout=30)
    rows = response.json()

    conn = psycopg2.connect(
        host="orchestack-postgres",
        dbname="data_warehouse",
        user="warehouse_admin",
        password="REPLACE_WITH_SECRET",   # use Airflow Variable in production
    )
    with conn.cursor() as cur:
        cur.execute("CREATE TABLE IF NOT EXISTS raw.api_data (id INT, payload JSONB)")
        for row in rows:
            cur.execute("INSERT INTO raw.api_data VALUES (%s, %s)", (row["id"], row))
    conn.commit()


profile_config = ProfileConfig(
    profile_name="orchestack_warehouse",
    target_name="prod",
    profile_mapping=PostgresUserPasswordProfileMapping(
        conn_id="orchestack_warehouse",
        profile_args={"schema": "public"},
    ),
)

with DAG(
    dag_id="path_b_python_dbt_tableau",
    start_date=datetime(2026, 1, 1),
    schedule="@daily",
    catchup=False,
    tags=["composition-pattern"],
) as dag:

    ingest = PythonVirtualenvOperator(
        task_id="python_ingest",
        python_callable=ingest_to_warehouse,
        requirements=["requests", "psycopg2-binary"],
        system_site_packages=False,
    )

    dbt_models = DbtDag(
        dag_id_prefix="dbt",
        project_config=ProjectConfig(Path("/opt/airflow/dbt-project")),
        profile_config=profile_config,
        execution_config=ExecutionConfig(dbt_executable_path="/home/airflow/.local/bin/dbt"),
    )

    # Tableau Server REST API: tell the published data source to refresh.
    refresh_tableau = HttpOperator(
        task_id="refresh_tableau",
        http_conn_id="tableau",                           # config in Airflow UI
        endpoint="api/3.x/sites/{{ var.value.tableau_site_id }}/datasources/{{ var.value.tableau_datasource_id }}/refresh",
        method="POST",
        headers={"X-Tableau-Auth": "{{ var.value.tableau_token }}"},
    )

    ingest >> dbt_models >> refresh_tableau

Why this pattern: Tableau lives on your own infrastructure (or Tableau Cloud) outside OrcheStack. You only need an HTTP connection to its REST API. The same pattern applies to Power BI's refresh API, Looker's content validator, Sigma's API — only the endpoint URL changes.

Path C: dlt → dbt → engineer queries warehouse

No BI tool yet. Data engineers query the warehouse directly via pgAdmin or psql. Ingestion uses dlt — a Python-first declarative loader popular with smaller engineering teams.

from datetime import datetime
from pathlib import Path

from airflow import DAG
from airflow.operators.python import PythonVirtualenvOperator

from cosmos import DbtDag, ProjectConfig, ProfileConfig, ExecutionConfig
from cosmos.profiles import PostgresUserPasswordProfileMapping


def dlt_load_github():
    # Runs in an ephemeral venv with dlt[postgres] installed.
    # dlt handles schema inference, idempotent loads, and write
    # disposition — much less code than raw psycopg2.
    import dlt

    @dlt.resource(name="github_stargazers", write_disposition="merge", primary_key="id")
    def stargazers():
        from dlt.sources.helpers import requests
        page = 1
        while True:
            response = requests.get(
                "https://api.github.com/repos/tripleaceme/orchestack-public/stargazers",
                params={"per_page": 100, "page": page},
                headers={"Accept": "application/vnd.github.star+json"},
            )
            data = response.json()
            if not data:
                return
            yield data
            page += 1

    pipeline = dlt.pipeline(
        pipeline_name="github_to_warehouse",
        destination="postgres",
        dataset_name="raw",
        progress="log",
    )
    pipeline.run(stargazers())


profile_config = ProfileConfig(
    profile_name="orchestack_warehouse",
    target_name="prod",
    profile_mapping=PostgresUserPasswordProfileMapping(
        conn_id="orchestack_warehouse",
        profile_args={"schema": "public"},
    ),
)

with DAG(
    dag_id="path_c_dlt_dbt_warehouse",
    start_date=datetime(2026, 1, 1),
    schedule="@daily",
    catchup=False,
    tags=["composition-pattern"],
) as dag:

    dlt_load = PythonVirtualenvOperator(
        task_id="dlt_load_github",
        python_callable=dlt_load_github,
        requirements=["dlt[postgres]>=1.0"],
        system_site_packages=False,
    )

    dbt_models = DbtDag(
        dag_id_prefix="dbt",
        project_config=ProjectConfig(Path("/opt/airflow/dbt-project")),
        profile_config=profile_config,
        execution_config=ExecutionConfig(dbt_executable_path="/home/airflow/.local/bin/dbt"),
    )

    dlt_load >> dbt_models

No third stage. Engineers query the resulting tables in pgAdmin or via psql. Adding BI later is a fourth task; the DAG above is already useful without it.

Setting dlt destination credentials: dlt reads PostgreSQL credentials from environment variables (DESTINATION__POSTGRES__CREDENTIALS__*). Set these in the Airflow Variable UI or via Airflow's environment-variable pass-through — never hardcode them in the DAG.

Set up your dbt project

All three paths assume your dbt project files exist at /opt/airflow/dbt-project inside the Airflow container. That path is a read-only mount of the same volume the dbt service uses (orchestack-dbt-repo), so you populate it from the dbt service side and it appears in Airflow.

Two ways to get your dbt project in there:

Set DBT_REPO_URL in your .env (or via the dashboard's Edit Config page). The dbt service clones the repo on next start.
Use the dbt terminal — open the dbt service tile in the dashboard, click "Open Terminal", and git clone directly into /usr/app/dbt.

Trigger and verify

Once your DAG file is in /opt/airflow/dags/ (mount that path from the host, set AIRFLOW_DAGS_REPO_URL to git-clone DAGs, or paste via the Airflow UI), Airflow discovers it within ~30 seconds.

Open the Airflow tile from the OrcheStack dashboard
Find your DAG in the list (tags help: filter by composition-pattern)
Trigger it manually first time (the ▶ button)
Open the graph view — for paths A/B/C, you'll see Cosmos's per-model expansion of dbt_models
If a model fails, click the specific failed task → re-run only that model after fixing

Once the DAG runs cleanly end-to-end, switch the schedule from manual to @daily (or your cadence) and Airflow takes it from there.

← Previous Step 3 — Configure services Next Architecture →