Declarative yaml-based airflow pipelines as Data Platform capability

Photo by Sen on Unsplash

full project in github: https://github.com/vanikkars/airflow-blue-print-project

Introduction

Every data platform team eventually reaches the same point: you have thousands ingestion DAGs, many working with identical source types and performing the same operations:

1. Read the source
2. Infer or validate the schema
3. Write the data and commit it to a catalog

Yet each pipeline is a slightly different Python implementation. At enterprise scale, this lack of standardization becomes difficult to maintain: changes that should happen once at the platform level instead require updates across hundreds of DAGs. Which leads to tedious errors and becomes a nightmare to maintain on scale.

There is also a usability problem. Sources such as CSV files and Google Sheets are often owned by business teams without the technical expertise to maintain Python and Airflow code. These people require more user-friendly interface, otherwise the platform risks to not meat the expected adoption rate.

In this article I will demonstrate how to standardize ingestion process via Yaml pipeline definitions and offer it as a platform capability.

I will use the Astronomer’s Blueprint library shared recently by Marc Lamberti for yaml-to-python translation.

A little spoiler:

---
dag_id: banking_users_to_iceberg
description: "Extract banking users tables to Iceberg in S3"
schedule: "@daily"
start_date: "2024-01-01"
catchup: false

steps:
  ingest_users_table:
    blueprint: postgres_to_iceberg_blueprint
    table_name: users
    schema_name: public
    iceberg_namespace: banking
    iceberg_table_name: users
    mode: overwrite
    postgres_conn_id: postgres_banking

That is an entire production-shaped pipeline: Postgres to Apache Iceberg

The yaml contains the high level configuration:

• name
• description
• schedule — “daily”
• start_date — “2024–01–01"
• whether the catchup is required

As well as the details for the source and target:

• table_name
• schema_name
• iceberg_namespace
• iceberg_table_name

All parameters are self-explanatory, to fill in all the values one does not need deep techincal knowledge.

This abstracaction from deep technical details of the Airflow dag definition creates a shift from: “the platform team writes pipelines” to “the platform team ships a capability, and users invoke it”.

Such approach simplifies the user experience and creates a clear border between the platform and the data pipeline owner.

The Python implementation is owned by the platform team.
The YAML pipeline definitions are owned by the teams that need the data.

Requirements

The goals I set for this project were:

1. Make the pipeline definition declarative — adding a table should not require writing or reviewing Python
2. Offer ingestion as a self-service capability, so non-engineers can create pipelines without a ticket
3. Keep the implementation typed and validated — a bad YAML key should fail at DAG-parse time, not at 3am in a task log
4. Make the blueprint reusable across any table, without per-table special cases
5. Standardise everything the user does not specify — retries, owner, catchup — so consistency is the default rather than a review comment
6. Run the entire stack locally, writing to a modern table format so the output is genuinely useful

Enough talking, let’s jump into the impementation.

Implementation

The project runs locally in Docker Compose: Airflow 2.11, Postgres for source data and Airflow metadata, a third backing the Iceberg REST catalog, and MinIO as S3-compatible storage. No cloud account required.

One blueprint, N pipelines

In large enterprises, it is common to operate thousands of ingestion pipelines against homogeneous source types: tables across PostgreSQL or MySQL databases, hundreds of CSV files in S3 buckets, or Google Sheets distributed across corporate drives.

While the number of pipelines can be large, their execution pattern is often nearly identical: extract, apply basic transformations, and load. What typically changes is configuration — the connection, source table or file, destination, and schedule.

Astronomer Blueprints provide a clean way to centralize this shared logic. The Blueprints library offers it as a Blueprint class.

Think of the Blueprint as a pipeline template: the platform team defines the pipeline’s behavior and the steps it must conduct.
For example, the PostgresToIcebergBlueprint defines the complete ingestion workflow — extraction, schema handling, transformation, writing, and catalog commits — in a single Python class.

class PostgresToIcebergBlueprint(Blueprint[PostgresToIcebergConfig]):
    """Blueprint for extracting Postgres tables to Iceberg format in S3."""

    def render(self, config: PostgresToIcebergConfig) -> TaskGroup:
        """Render the task group for Postgres to Iceberg ingestion."""

        iceberg_table_name = config.iceberg_table_name or config.table_name

        with TaskGroup(group_id=f"ingest_{config.table_name}") as task_group:

            @task(task_id="extract_postgres_schema")
            def extract_postgres_schema(**context) -> dict[str, Any]:
                """Extract table schema from Postgres."""
                ...

                return {"columns": columns, "table_name": config.table_name}

            @task(task_id="extract_postgres_data")
            def extract_postgres_data(schema_info: dict[str, Any], **context) -> str:
                """Extract data from Postgres and save as Parquet."""
                ...

                print(f"Extracted {len(df)} rows from {config.table_name}")
                return temp_file.name

            @task(task_id="create_iceberg_table")
            def create_iceberg_table(
                schema_info: dict[str, Any], **context
            ) -> dict[str, str]:
                """Create or update Iceberg table."""
               ...
                return {
                    "table_identifier": table_identifier,
                    "warehouse": catalog_config["warehouse"],
                }


            @task(task_id="load_to_iceberg")
            def load_to_iceberg(
                parquet_file: str, iceberg_info: dict[str, str], **context
            ) -> dict[str, Any]:
                """Load Parquet data into Iceberg table."""
                ...
                return {
                    "table": iceberg_info["table_identifier"],
                    "rows_loaded": row_count,
                    "mode": config.mode,
                }


            # Define task dependencies
            schema_info = extract_postgres_schema()
            parquet_file = extract_postgres_data(schema_info)
            iceberg_info = create_iceberg_table(schema_info)
            result = load_to_iceberg(parquet_file, iceberg_info)

        return task_group

The project ships one blueprint implementation — postgres_to_iceberg.py, roughly 300 lines.

Get Vanik Karslian’s stories in your inbox

Join Medium for free to get updates from this writer.

Remember me for faster sign in

Now, we can define two piplines that would share the common logic, defined in PostgresToIcebergBlueprint:

the one ingests the transactions:

steps:
  ingest_transactions_table:
    blueprint: postgres_to_iceberg_blueprint
    table_name: transactions
    schema_name: public
    iceberg_namespace: banking
    iceberg_table_name: transactions
    s3_bucket: iceberg-warehouse
    s3_prefix: warehouse
    mode: overwrite
    batch_size: 100
    postgres_conn_id: postgres_banking

the other — the users table:

steps:
  ingest_users_table:
    blueprint: postgres_to_iceberg_blueprint
    table_name: users
    schema_name: public
    iceberg_namespace: banking
    iceberg_table_name: users
    s3_bucket: iceberg-warehouse
    s3_prefix: warehouse
    mode: overwrite
    batch_size: 100
    postgres_conn_id: postgres_banking

As you can see the creation of a new pipeline with users just required to change the input and output parameters, and that’s it.

Everything else — the schema extraction, the Postgres-to-Iceberg type mapping, the batched read, the Parquet staging, the catalog commit, the timestamp fixes, the retry policy — is inherited. Not copied. Inherited.

Now, in order to integrate another source table into iceberg destination one need to create a new yaml file.

This is the difference between a template and a capability:

The YAML file is the interface

If the blueprint is the capability developed by the data platform team and offered to the end customers. Thus, the YAML file plays a role of a public inteface. Given that it’s a public intrface we should also validate the user input. Here is when the Pydantic model plays a crucial role. The pydantic allows provides us the validation out of the box- which fields are required, what are the expected values and types:

class PostgresToIcebergConfig(BaseModel):
    """Configuration for Postgres to Iceberg ingestion."""

    table_name: str = Field(..., description="Name of the Postgres table to ingest")
    schema_name: str = Field(default="public", description="Postgres schema name")
    iceberg_namespace: str = Field(
        default="default", description="Iceberg namespace/database"
    )
    mode: str = Field(default="overwrite", description="Write mode: 'overwrite' or 'append'")
    batch_size: int = Field(default=10000, description="Number of rows to fetch per batch")
    postgres_conn_id: str = Field(
        default="postgres_banking", description="Airflow connection ID for source Postgres"
    )

in order to invoke the blue-print, one need to merely define it’s name in the pipeline’s yaml:

steps:
  ingest_loans:
    blueprint: postgres_to_iceberg_blueprint
    table_name: loans

As a bonus of using Pydantic — each yaml file gets validated before anything runs. Pydantic checks the config at DAG-parse time, thus any mistakes or typos in the yaml file, such as iceberg_namespcae will be captured immediately.

The blast radius is bounded. A YAML author chooses which table, where it lands, and how often and the validation comes out of the box.

The ownership split ends up clean:

essentially, the platofrm team owns the pipeline’s implementation details and the users — their pipelines defined in yaml.

Standardisation and code duplicaiton

Let’s now see how the blue prints help to avoid code duplication. Imagine, we want to provide reasonable default values for certain pipeline parameters, such as number of retries or retry delay. In order to implement it, we need to define the ProjectDagArgsConfig class:

class ProjectDagArgsConfig(BaseModel):
    schedule: str | None = None
    start_date: str | None = None
    catchup: bool = False
    owner: str = "data-engineering"
    retries: int = 2
    retry_delay_minutes: int = 5


class ProjectDagArgs(BlueprintDagArgs[ProjectDagArgsConfig], default=True):
    """DAG arguments shared by every pipeline in this project."""

    def render(self, config: ProjectDagArgsConfig) -> dict[str, Any]:
        return {
            "catchup": config.catchup,
            "default_args": {
                "owner": config.owner,
                "retries": config.retries,
                "retry_delay": timedelta(minutes=config.retry_delay_minutes),
            },
            ...
        }

default=True applies this to every DAG in the project. Every pipeline retries twice, carries an owner, and has catchup off — not because each author remembered, but because those are the defaults. Nobody writing YAML needs to know what catchup does to avoid being burned by it.

Three consequences: platform-wide policy becomes a one-line edit rather than forty. The schema acts as a whitelist — users cannot set an arbitrary Airflow parameter, so they cannot set a harmful one. And variance disappears: templating removes duplication, this removes inconsistency, and only the second is what people mean by “standardised pipelines.”

Rendering the task group

A Blueprint is the config model above plus a render() method turning it into tasks. It returns a TaskGroup, so each YAML step becomes a grouped unit in the Airflow UI:

class PostgresToIcebergBlueprint(Blueprint[PostgresToIcebergConfig]):
    """Blueprint for extracting Postgres tables to Iceberg format in S3."""

    def render(self, config: PostgresToIcebergConfig) -> TaskGroup:
        iceberg_table_name = config.iceberg_table_name or config.table_name

        with TaskGroup(group_id=f"ingest_{config.table_name}") as task_group:

            @task(task_id="extract_postgres_schema")
            def extract_postgres_schema(**context) -> dict[str, Any]: ...
            # extract_postgres_data / create_iceberg_table / load_to_iceberg
            # follow the same shape, each reading `config` from the closure

            schema_info = extract_postgres_schema()
            parquet_file = extract_postgres_data(schema_info)
            iceberg_info = create_iceberg_table(schema_info)
            result = load_to_iceberg(parquet_file, iceberg_info)

        return task_group

config is captured by closure, so each task reads config.table_name or config.mode directly — no XCom plumbing for configuration, no params dictionary to get wrong.

The loader: how YAML becomes DAGs

Airflow only executes Python files; it will never turn a bare .yaml into a DAG. One twelve-line module bridges the gap, and is never touched again:

build_all_airflow_dags(
    search_path=_DAGS_DIR,
    register_globals=globals(),
    bp_registry=BlueprintRegistry(template_dirs=[_BLUEPRINTS_DIR, _DAGS_DIR]),
)

It discovers every *.dag.yaml, resolves each step's blueprint: key against the registry, validates the config against the Pydantic model, and injects the DAG objects into module globals for the scheduler.

This is what makes the whole thing self-service: a new pipeline requires no registration step. Drop a file into dags/, and it exists — no list to append to, nothing a new user can forget.

The unglamorous fixes that reusability pays for

Two type problems bit me, and they are the clearest argument for centralising an implementation.

1. Nanosecond timestamps.
   pandas produces timestamp[ns]; Iceberg tops out at microseconds, so an untouched frame fails with UnsupportedPyArrowTypeException:

def _downcast_ns_timestamps(table: "pa.Table") -> "pa.Table":
    """Cast any nanosecond timestamp column to microseconds."""
    import pyarrow as pa

    fields = []
    changed = False
    for field in table.schema:
        if pa.types.is_timestamp(field.type) and field.type.unit == "ns":
            fields.append(field.with_type(pa.timestamp("us", tz=field.type.tz)))
            changed = True
        else:
            fields.append(field)

    if not changed:
        return table
    return table.cast(pa.schema(fields))

2. Schema drift.
pandas widens ints to int64 and marks every column nullable, which Iceberg rejects for int32 or NOT NULL columns. The fix is one line — arrow_table.cast(table.schema().as_arrow()) — casting to the target table's own Arrow schema before writing.

Neither fix is interesting, and that is the point. Nobody writing YAML will encounter either, or need to learn the words “PyArrow schema coercion.” The platform absorbed both, permanently, for every pipeline that will ever exist.

This is the real argument for blueprints. Not that YAML is prettier than Python — that a hard-won fix has exactly one home, and the people creating pipelines never have to learn the problem it solves.

Summary

Let’s some up what has been implemented.

1. We have created a Blueprint class PostgresToIcebergBlueprint that extracts data from postgres and ingests into Iceberg
2. We defined ProjectDagArgsConfig to store the default params for all the dags in the project
3. We provided a yaml-based public interface for our users

And lust but not least, we introduce a significant change in the approach- a clear separation between platform and users ownership: