# Ingest with dlt, transform with dbt (/pipelines/dlt-and-dbt) Load API data into Iceberg with dlt and transform it with dbt on Trino, from an Airflow task. [dlt](https://dlthub.com/docs/intro) and [dbt](https://docs.getdbt.com/docs/introduction) 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](/pipelines/authoring). ## 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](https://dlthub.com/docs/dlt-ecosystem/verified-sources/rest_api/basic) docs for the full configuration. ```python title="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](/pipelines/connections). ### 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. ```python title="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](/pipelines/authoring#a-pipeline-is-python). 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](/pipelines/authoring#files-get_store) and [recording what landed](/pipelines/authoring#record-what-landed-get_index). ## 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](/getting-started/concepts#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: ```yaml title="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. ```yaml title="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](https://docs.getdbt.com/docs/local/connect-data-platform/trino-setup) 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: ```python title="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](#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, '')` 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.