Ingest with dlt, transform with dbt

Load API data into Iceberg with dlt and transform it with dbt on Trino, from an Airflow task.

dlt and dbt are both preinstalled for pipelines, so a task can import them directly. This page covers what is specific to Databaas: where the credentials come from, how data lands in the catalog, and how dbt reaches Trino. It assumes you know how a pipeline is written.

Ingest with dlt

A @dlt.source describes an API declaratively (base URL, resources, pagination, authentication), and dlt’s REST API helper handles the requests. See dlt’s REST API source docs for the full configuration.

airflow/example_source_pipeline/source.py

import dlt
from dlt.sources.rest_api import RESTAPIConfig, rest_api_resources


@dlt.source(name="example_source")
def example_source(api_key: str = dlt.secrets.value):
    config: RESTAPIConfig = {
        "client": {
            "base_url": "https://api.example.com/v1/",
            "auth": {
                "type": "api_key",
                "name": "Authorization",
                "api_key": api_key,
                "location": "header",
            },
        },
        "resources": [{"name": "records", "endpoint": {"path": "records"}}],
    }
    yield from rest_api_resources(config)


def fetch_records(api_key: str) -> list[dict]:
    records = list(example_source(api_key=api_key).resources["records"])
    if not records:
        raise RuntimeError("example_source returned no records")
    return records

Pass the API key in as an argument. It comes from an Airflow connection, not from a dlt secrets file: a pipeline task has no .dlt/ directory. See connections and variables.

Write the records into an Iceberg table

When the source returns structured records, write them into Iceberg through get_catalog(). pyarrow infers the schema from the records, nested structures included.

dags/example_source_ingest.py

import pendulum
from airflow.sdk import dag, task
from databaas_utils.airflow import scoped
from databaas_utils.airflow.target import target

NAMESPACE = scoped("example_source")


@dag(
    dag_id=target.dag_id("example_source_ingest"),
    schedule="@daily",
    start_date=pendulum.datetime(2026, 9, 1, tz="UTC"),
    catchup=False,
    is_paused_upon_creation=target.paused,
    tags=target.tags("example_source"),
)
def example_source_ingest():
    @task(retries=3)
    def ingest_records() -> int:
        from datetime import datetime, timezone

        import pyarrow as pa
        from airflow.sdk import BaseHook
        from databaas_utils.airflow import get_catalog
        from example_source_pipeline.source import fetch_records

        api_key = BaseHook.get_connection("source_opendata").password
        fetched_at = datetime.now(timezone.utc).isoformat()
        rows = [{"fetched_at": fetched_at, **r} for r in fetch_records(api_key)]

        catalog = get_catalog()
        catalog.create_namespace_if_not_exists(NAMESPACE)
        data = pa.Table.from_pylist(rows)
        table = catalog.create_table_if_not_exists(f"{NAMESPACE}.records", schema=data.schema)
        table.append(data)
        return data.num_rows

    ingest_records()


example_source_ingest()

This appends: each run adds one more batch of history. If the source hands back a full snapshot instead, overwrite, or overwrite the run’s own slice with a filter as in authoring pipelines. Match the write to what the source gives you.

Land a file instead

When the payload is not tabular yet (a NetCDF or GRIB file, a zip), land it first and parse it in a separate task. Store it with get_store().put(...), record it with get_index().record(...), and return only the URI and key. See files and recording what landed.

Transform with dbt

dbt runs against Trino, reading and writing Iceberg tables through the catalog. The project itself is a standard dbt-trino project: dbt_project.yml, profiles.yml, and a models/ tree with sources.yml and your models. Keep it in your pipeline repository next to the DAG.

The catalog is always `iceberg`

Whatever your pipeline is called, the Trino catalog in profiles.yml (catalog:) and in sources.yml (database:) is the literal iceberg. What dbt-trino calls a schema is a Databaas namespace.

Configure the profile

Your Databaas team provides the connection values for your pipeline’s dbt profile: method, host, port, http_scheme and user. dbt then queries Trino as your pipeline’s own application identity, so it reads and writes only what that identity is granted. Read the values from environment variables so nothing is hardcoded, and read the catalog and output namespace the same way:

profiles.yml

example_project:
  target: trino
  outputs:
    trino:
      type: trino
      method: "{{ env_var('TRINO_METHOD') }}"
      user: "{{ env_var('TRINO_USER') }}"
      host: "{{ env_var('TRINO_HOST') }}"
      port: "{{ env_var('DBT_TRINO_PORT') | as_number }}"
      http_scheme: "{{ env_var('TRINO_HTTP_SCHEME') }}"
      catalog: "{{ env_var('TRINO_CATALOG', 'iceberg') }}"
      schema: "{{ env_var('TRINO_NAMESPACE') }}"
      threads: 4

Do not name the port variable TRINO_PORT: that name can already be set in the task environment, and it would override yours.

models/sources.yml

sources:
  - name: example_source
    database: "{{ env_var('TRINO_CATALOG', 'iceberg') }}"
    schema: "{{ env_var('SOURCE_NAMESPACE') }}"
    tables:
      - name: records

See dbt’s Trino profile reference for every other field.

Scope namespaces per deployment

A branch deploy must write to its own namespaces, like any other pipeline. Compute scoped names in the DAG file and export them before invoking dbt:

dags/example_transform.py (excerpt)

from pathlib import Path

from airflow.sdk import task
from databaas_utils.airflow import scoped

DBT_PROJECT_DIR = str(Path(__file__).resolve().parents[1] / "example_dbt")
NAMESPACES = {
    "SOURCE_NAMESPACE": scoped("example_source"),
    "TRINO_NAMESPACE": scoped("example_transformed"),
}


@task
def dbt_run(project_dir: str, namespaces: dict[str, str]) -> None:
    import os

    from dbt.cli.main import dbtRunner

    os.environ["TRINO_CATALOG"] = "iceberg"
    os.environ.update(namespaces)

    result = dbtRunner().invoke(
        [
            "run",
            "--project-dir", project_dir,
            "--profiles-dir", project_dir,
            "--select", "+my_model",
        ]
    )
    if not result.success:
        raise RuntimeError(f"dbt run failed: {result.exception or result.result}")

Call it inside your DAG as dbt_run(DBT_PROJECT_DIR, NAMESPACES). The task also needs the profile values from configure the profile in its environment.

Run the whole project as one task with dbt-core’s dbtRunner, and fail the task when dbt fails.
Select +my_model, not my_model. The leading + builds every upstream model too; without it, the first run against a fresh branch namespace fails with TABLE_NOT_FOUND.

Trino and Iceberg SQL differences

dbt compiles your SQL; Trino runs it against Iceberg. Each of these shows up as a clear error on the first run and is a one-line fix:

Timestamp precision. A bare CAST(... AS timestamp) has precision 3, and Iceberg accepts only 6. Cast to timestamp(6), or the write fails with NOT_SUPPORTED.
Parsing timestamp strings. An ISO 8601 string without a timezone or seconds can fail a direct cast with INVALID_CAST_ARGUMENT. Use date_parse(value, '<format>') first, then cast.
Unnesting parallel arrays. CROSS JOIN UNNEST(arr1, arr2) AS t(col1, col2) unnests several arrays together, positionally, into one row per element.

Ingest with dlt, transform with dbt

On this page