Before data can be transformed into insights or serve as input for models, the data first has to be ingested on the Data Platform. In this blog post we will describe how data can be loaded into the Databaas Catalog with the use of the dlt package. Airflow can run any Python-based framework, so there are many ways to ingest data with an Airflow task. In this post we use dlt, because it’s easy to use, as you’ll see below.

Before you start: know your ingestion pattern

Every source is a little different, before writing any code, it’s worth answering:

  • Connection type: does the source hand back structured data from a single call (eg. a REST API returning JSON), or do get back an object (eg. a file)?
  • Pagination: does one call return all the data, or does the source have multiple pages with results?
  • Authentication: what does the source need to authenticate?
  • Append or overwrite: does each fetch give you a full, current snapshot (overwrite what’s there), or only a slice (append to it)?
  • Schedule: how often does the source actually publish new data? A schedule that polls more often just re-fetches the same data and less often risks missing data.
  • Idempotency: what happens if this task runs twice, on a retry, a backfill, or an overlapping run? Whatever you write should be safe to run again, not just safe to run once.

Each answer helps to clarify how the ingestion task should be set up.

What is dlt

dlt (data load tool) is an open-source Python package for pulling data from many different sources (REST APIs, files, databases) into structured, typed datasets. Pagination, authentication and schema handling are provided in the config rather than something you have to write yourself per source: a base URL, a list of resources, a paginator type are here the inputs.

Below we show how dlt can be used for ingesting a structured response and the pattern of ingesting a file onto Databaas. The examples are both also available with a working example in the repo databaas-dlt-dbt-example, a repo ingesting marine wave-height forecasts, wind forecasts, and KNMI weather station observations. Treat the repo as a reference, for your source the answers to the questions above may differ and result in a different setup.

We’ll only cover the ingestion half of the example repo here. The transformation layer, which uses dbt to combine these sources once they’ve landed in the Databaas Catalog, is described in another blog post.

Pattern one: straight into a table

The straightforward case: the source already returns structured data, so dlt’s rest_api source can fetch the data and the result can be written straight into an Iceberg table. Here’s how the example repo’s marine forecast source does it with a RESTAPIConfig, one call to an endpoint returning an hourly forecast payload:

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

@dlt.source(name="marine_forecast")
def marine_forecast_source(location: Location, forecast_days: int):
    config: RESTAPIConfig = {
        "client": {"base_url": "https://marine-api.open-meteo.com/v1/"},
        "resources": [
            {
                "name": "forecast",
                "endpoint": {
                    "path": "marine",
                    "params": {
                        "latitude": location.latitude,
                        "longitude": location.longitude,
                        "hourly": ",".join(HOURLY_VARIABLES),
                        "forecast_days": forecast_days,
                        "timezone": "UTC",
                    },
                    # The API returns one JSON object, not a list of them --
                    # "$" yields the whole response as a single record.
                    "data_selector": "$",
                    "paginator": {"type": "single_page"},
                },
            },
        ],
    }
    yield from rest_api_resources(config)

Pagination is config here: this particular API returns everything in one response, so the paginator is pinned to single_page rather than left to auto-detection. An API that pages through results would set a different paginator type instead. See dlt’s own REST API docs for the full list. The data returned from dlt, is written here to the Catalog using the pyiceberg function create_table_if_not_exists with the Databaas catalog.
NOTE, the catalog is created based on the scoped namespace to ensure the data for dev and prod land in a different namespace.

def write_table(rows: list[dict]) -> int:
    cat, settings = catalog()
    ns = settings.namespace
    cat.create_namespace_if_not_exists(ns)

    data = pa.Table.from_pylist(rows)
    table = cat.create_table_if_not_exists(f"{ns}.marine_table", schema=data.schema)
    table.append(data)
    return data.num_rows

The schema is not declared seperatly, pyarrow infers it from the data dlt returned, and the table adapts as the source’s own fields change. So the pattern for structured data is: fetch with dlt, hand the records to pyiceberg.

Pattern two: land it as a file first

Some sources don’t return structured data. The example repo’s KNMI weather station source is an Open Data API file API it lists the files in a dataset, provides a short-lived signed download URL for one file and then fetches the actual bytes from that URL. That’s a three-step flow rest_api_resources a declarative dlt config can’t express, and the payload itself (a NetCDF binary) isn’t in a structure dlt can normalize on its own.

The fix is to split the task in two. Listing files is structured (filename, size, last-modified), so that part is still a @dlt.resource:

@dlt.resource(name="recent_files")
def recent_files(api_key: str, files_per_run: int):
    resp = requests.get(
        f"{BASE_URL}/datasets/{DATASET}/versions/{VERSION}/files",
        params={"maxKeys": files_per_run, "sorting": "desc", "orderBy": "lastModified"},
        headers={"Authorization": api_key},
    )
    resp.raise_for_status()
    yield from resp.json()["files"]

Downloading one file’s bytes isn’t structured data, so that’s a plain function, since dlt has nothing to normalize there:

def download(api_key: str, filename: str) -> bytes:
    resp = requests.get(download_url(api_key, filename), timeout=60)
    resp.raise_for_status()
    return resp.content

The first step lands each downloaded file raw in object storage first, and records it in an index. After that the second step fetches each file and parses the row(s) it actually needs before appending it into an Iceberg table. Landing the raw file before parsing it means a parsing bug can be fixed and re-run against files already downloaded, instead of re-hitting the API.

Authentication using Airflow connections

KNMI’s API authenticates with and API key. The API key is stored in an Airflow connection of type Generic in the password field, not in the DAG file itself:

def api_key_from_connection(conn_id: str = "KNMI_REGISTERED_API_KEY") -> str:
    from airflow.sdk import BaseHook

    conn = BaseHook.get_connection(conn_id)
    value = conn.password
    if not value or not value.strip():
        raise RuntimeError(f"connection {conn_id!r} has no password (the KNMI API key)")
    return value.strip()

Airflow has many different types of connections. Generic is the right connection type here as we are only storing the API Key with no host/login/extra structure that would justify a more specific type. As a rule of thumb, anything that authenticates a source belongs in a connection. An Airflow Variable is for non-secret pipeline state, not credentials.

Orchestration with Airflow

Each ingestion step runs as an Airflow task, a Python function decorated with @task. dlt is already installed in the Databaas runtime, along with other packages; see the Databaas docs for the full list. Only if you need a package that isn’t in the runtime yet do you need a @task.virtualenv task instead.

@task(retries=3, retry_delay=timedelta(minutes=2))
def ingest_marine_forecast() -> int:
    from marine_pipeline import lakehouse, marine
    from marine_pipeline.models import EUROPLATFORM_51_9978_3_2750

    payload = marine.fetch_forecast(EUROPLATFORM_51_9978_3_2750, forecast_days=7)
    row = {
        "location": EUROPLATFORM_51_9978_3_2750.name,
        "fetched_at": datetime.now(timezone.utc).isoformat(),
        **payload,
    }
    return lakehouse.write_table([row])

Once a task like this succeeds, the DAG can emit an Asset, a named signal other DAGs can schedule against instead of guessing at a fixed time gap:

from airflow.sdk import Asset, task

MARINE_FORECAST_ASSET = Asset(name="marine_forecast_ingested")

@task(
    ...,
    outlets=[MARINE_FORECAST_ASSET],
)
def ingest_marine_forecast() -> int:
    ...

By using outlets on the task decorator, Airflow emits MARINE_FORECAST_ASSET when this task succeeds, and any other DAG can be scheduled by the Asset name. No polling, no cron required.

An Asset can then be used to trigger for example a transformation DAG. Each source keeps its own schedule and ingests independently and a shared transformation DAG can wait on all of their Assets at once and run the moment any one of them lands: Both ingestion patterns emit an Asset on success, feeding a shared transformation DAG

In a future post we will show how you can use, for example, dbt in that transformation DAG to combine these ingested sources into insights on Databaas.

Build it yourself

The Databaas docs describe the Databaas platform and contain additional information on working with Airflow pipelines. databaas-dlt-dbt-demo contains the working pipeline with tests used in this blog post as the example.