Metadata-Driven Engineering at Scale with Datus AI

One reliable pipeline is an achievement. Fifty reliable pipelines are a different problem. At that point, the challenge is no longer writing code. It is keeping every pipeline consistent, documented, and safe to change.

In last blog, we showed how to engineer reliable data pipelines with Datus AI. Earlier, we covered why analytics projects fail, the unified operating model, AI-assisted planning, and the journey from business questions to architecture.

This article goes one level deeper. It explains metadata-driven engineering, why mature organizations adopt it, and how Datus generates the artifacts that make it work.

What Is Metadata-Driven Engineering?

Metadata-driven engineering is an approach where metadata describes the pipeline, and a generic engine builds and runs it. Instead of writing a new script for every dataset, teams describe each dataset in a structured file. The engine reads that file and does the rest.

Consider the traditional approach. An engineer writes one ingestion script per source. Another engineer copies it and changes a few lines. Soon, dozens of near-identical scripts exist. Each one drifts in its own direction.

In metadata-driven engineering, the logic is written once. Only the description changes.

Think of it as a recipe book. The recipes are metadata, and the kitchen is the engine. Adding a dish means adding a recipe, not building a new kitchen.

Here is the same idea in code. The first code snippet shows the traditional style, and the latter code snippet side shows the metadata-driven style:

# Traditional: one script per source, logic copied each time
def load_attendance():
    df = read_file("attendance_2024.txt")
    df = df.rename(columns={"CDS": "cds_code"})
    write("raw.attendance", df)

def load_suspensions():
    df = read_file("suspension_2024.txt")
    df = df.rename(columns={"CDS": "cds_code"})
    write("raw.suspensions", df)

# Metadata-driven: one function, many descriptions
def ingest(source: dict, school_year: str):
    df = read_file(source["location"].format(school_year=school_year))
    df = df.rename(columns=source["column_map"])
    write(source["target_raw"], df)

The second version scales without copying code. That difference matters most when the number of datasets grows.

Why Mature Organizations Adopt It?

Small teams can survive with scripts. Mature organizations cannot. As the platform grows, the same pain points appear again and again.

Three reasons stand out.

First, consistency, when every pipeline runs through the same engine, every pipeline handles retries, logging, and failures the same way.

Second, governance, metadata records owners, classifications, and definitions. Therefore, controls apply automatically instead of depending on memory.

Third, speed with safety, teams move faster because they change descriptions, not code. Meanwhile, reviews stay simple because a reviewer reads a short file.

Continuing our school scenario, the school board started with a few datasets. Now it tracks assessment, attendance, suspension, graduation, and funding data across many years. Without metadata, that growth would mean dozens of hand-built pipelines.

Metadata Artifacts Generated by Datus

Metadata-driven engineering only works if good metadata exists. Writing it by hand is slow, and that is where many teams give up. Datus solves this by generating the artifacts from the work it already does.

Here are three examples. The following snippets are illustrative, so adapt the structure to your own standards.

A source registry entry tells the engine how to ingest a file:

source: chronic_absenteeism
domain: Attendance
format: delimited
delimiter: "\t"
location: "landing/attendance/{school_year}/chronic_absenteeism.txt"
expected_columns: [CDS_CODE, StudentGroup, ChronicAbsentCount, Enrollment]
column_map:
  CDS_CODE: cds_code
  StudentGroup: student_group
  ChronicAbsentCount: chronic_absent_count
  Enrollment: eligible_enrollment
target_raw: raw.chronic_absenteeism
owner: Research and Accountability team

A mapping specification describes how raw data becomes canonical data:

target: cleansed.attendance
source: raw.chronic_absenteeism
columns:
  - {name: cds_code, source: cds_code, type: STRING}
  - {name: student_group, source: student_group, type: STRING}
  - {name: chronic_absent_count, source: chronic_absent_count, type: INT, suppression_marker: "*"}
  - {name: eligible_enrollment, source: eligible_enrollment, type: INT}

A quality rule file declares the checks:

dataset: curated.fact_attendance
rules:
  - {id: Q1, type: unique, columns: [cds_code, student_group, school_year]}
  - {id: Q2, type: not_null, columns: [cds_code, school_year]}
  - {id: Q3, type: expression, expression: "chronic_absent_count <= eligible_enrollment"}

Together, these files describe a complete pipeline without a single line of pipeline code. Datus proposes them, and people review and approve them in Git, just like code.

Notice the loop. Run results flow back into Datus, which then proposes metadata updates. As a result, metadata stays current instead of decaying.

Pipeline Automation Examples

Now let’s see the engine work. The next four examples show how metadata turns into running pipelines.

Example 1: Generate SQL from a mapping

Instead of writing each transformation by hand, a small generator reads the mapping specification:

def build_cleanse_sql(spec: dict) -> str:
    cols = []
    for c in spec["columns"]:
        expr = c["source"]
        marker = c.get("suppression_marker")
        if marker:
            expr = (f"CASE WHEN {expr} = '{marker}' THEN NULL "
                    f"ELSE CAST({expr} AS {c['type']}) END")
        elif c["type"] != "STRING":
            expr = f"CAST({expr} AS {c['type']})"
        cols.append(f"    {expr} AS {c['name']}")
    body = ",\n".join(cols)
    return f"CREATE OR REPLACE VIEW {spec['target']} AS\nSELECT\n{body}\nFROM {spec['source']};"

Run against the mapping shown earlier, it produces:

CREATE OR REPLACE VIEW cleansed.attendance AS
SELECT
    cds_code AS cds_code,
    student_group AS student_group,
    CASE WHEN chronic_absent_count = '*' THEN NULL ELSE CAST(chronic_absent_count AS INT) END AS chronic_absent_count,
    CAST(eligible_enrollment AS INT) AS eligible_enrollment
FROM raw.chronic_absenteeism;

The suppression rule from the last blog is now a metadata setting. Every dataset with suppressed values gets the same correct treatment.

Example 2: Turn quality rules into checks

The same idea applies to quality. Each rule type maps to one query, and a passing rule returns zero rows:

def rule_to_sql(table: str, rule: dict) -> str:
    if rule["type"] == "unique":
        cols = ", ".join(rule["columns"])
        return (f"SELECT {cols}, COUNT(*) AS n FROM {table} "
                f"GROUP BY {cols} HAVING COUNT(*) > 1")
    if rule["type"] == "not_null":
        cond = " OR ".join(f"{c} IS NULL" for c in rule["columns"])
        return f"SELECT * FROM {table} WHERE {cond}"
    if rule["type"] == "expression":
        return f"SELECT * FROM {table} WHERE NOT ({rule['expression']})"
    raise ValueError(f"Unknown rule type: {rule['type']}")

Adding a new rule means adding one line to a YAML file. No developer needs to write a new test.

Example 3: Generate one pipeline per dataset

Orchestration can also be driven by metadata. Using Airflow-style syntax, one factory function builds a pipeline for every configuration file:

from datetime import datetime
from airflow.decorators import dag, task

def build_dag(cfg: dict):
    @dag(
        dag_id=f"load_{cfg['name']}",
        schedule=cfg["schedule"],
        start_date=datetime(2026, 1, 1),
        catchup=False,
        default_args={"retries": cfg.get("retries", 3)},
    )
    def pipeline():
        @task
        def ingest():
            run_ingest(cfg)

        @task
        def standardize():
            run_standardize(cfg)

        @task
        def check_quality():
            run_quality(cfg)

        ingest() >> standardize() >> check_quality()

    return pipeline()

for cfg in load_pipeline_configs():
    globals()[f"load_{cfg['name']}"] = build_dag(cfg)

Here, run_ingest, run_standardize, run_quality, and load_pipeline_configs are helper functions that read the metadata files. Datus can generate an equivalent for the orchestrator your team already uses.

Example 4: Onboard a new dataset

Now suppose the board wants to add suspension data. In a metadata-driven setup, the process is short.

The team adds a registry entry. Datus drafts the mapping, contract, and quality rules. A reviewer approves the pull request. Automated checks validate the files, and the pipeline appears on its own. No new pipeline code is written.

Operational Benefits

The real payoff comes after launch. Teams that adopt metadata-driven engineering notice benefits in daily operations.

Impact analysis becomes simple. Because lineage lives in metadata, a short script can answer, “What breaks if this table changes?”

from collections import defaultdict

def impact(specs: list[dict], changed: str) -> list[str]:
    graph = defaultdict(list)
    for s in specs:
        for upstream in s["lineage"]["upstream"]:
            graph[upstream].append(s["dataset"])
    seen, stack = set(), [changed]
    while stack:
        node = stack.pop()
        for child in graph[node]:
            if child not in seen:
                seen.add(child)
                stack.append(child)
    return sorted(seen)

# impact(specs, "raw.chronic_absenteeism")
# -> ["curated.fact_attendance", "semantic.school_card"]

Beyond impact analysis, teams see several other gains:

• Faster onboarding, new datasets need descriptions, not custom code.
• Fewer defects, tested logic is reused instead of rewritten.
• Consistent quality, every dataset follows the same rule engine.
• Audit readiness, contracts, owners, and decisions are versioned and traceable.
• Easier handover, new engineers read the metadata to understand the platform.
• Safer change, reviewers see exactly what changed in a pull request.

Pitfalls to Avoid

This approach is powerful, but it has traps.

• Keep metadata in Git, unversioned metadata becomes untrusted metadata.
• Review what Datus generates, generated does not mean approved.
• Avoid an engine that does everything, keep the engine simple, and handle true exceptions with custom code.
• Validate metadata automatically, check structure, required fields, and references in CI.
• Assign owners, every file needs a person accountable for it.

In Summary

Metadata-driven engineering replaces repeated, hand-built pipelines with one engine and many descriptions. It gives mature organizations consistency, governance, and speed at the same time.

Datus makes the approach practical. It generates source registries, mappings, contracts, quality rules, pipeline configurations, and lineage. People review that metadata, the engine runs it, and results flow back to keep it current. As a result, teams spend less time maintaining pipelines and more time delivering insight.