Designing a Data Pipeline for Massive Tables That Scales with Minimal Code

This article was written using AI.

Overview

In data engineering practice, managing pipelines becomes unmanageable as the number of tables to ingest grows to hundreds or thousands. This article explains how to use a Config-driven pipeline design approach to scale a large number of tables on Databricks with minimal code.

---

The Common Problem: Code Grows Every Time a Table Is Added

When pipelines are written per table, the following problems accumulate.

Similar code appears everywhere

# Pipeline for orders
def ingest_orders():
    df = spark.read.format("jdbc").option("url", ...).option("query", "SELECT * FROM orders").load()
    df.write.format("delta").mode("append").saveAsTable("bronze.orders")

# Pipeline for contacts (almost identical)
def ingest_contacts():
    df = spark.read.format("jdbc").option("url", ...).option("query", "SELECT * FROM contacts").load()
    df.write.format("delta").mode("append").saveAsTable("bronze.contacts")

# Pipeline for events (also almost identical)
def ingest_events():
    ...

If there are 10 tables, you end up with 10 nearly identical functions. When the connection URL changes, you have to fix it in 10 places, and missed fixes become a source of bugs.

A Python file has to be written every time a new table is added

Even to add a single table, the full cycle of creating a new Python file, going through review, and deploying it happens every time. The operational cost is high, and mistakes are easy to make.

Quality checks and error handling are implemented differently across each pipeline

When you want to fix quality check logic, you have to go and update every single pipeline. The larger the team, the more inconsistency in implementation becomes apparent.

---

Prerequisite 1: What Are Bronze, Silver, and Gold Layers?

Before getting into the solution, let me clarify the layer concept used in this article.

In data lake architecture, data is managed in three layers based on the degree of processing applied.

Source (DB/API/Files)
     │
     ▼
Bronze Layer: Raw data stored as-is. No transformation.
     │
     ▼
Silver Layer: Cleansed and joined data. Ready for analysis.
     │
     ▼
Gold Layer: Data aggregated with business logic. Directly connected to reports and BI.

Separating into layers makes it clear "how much the data has been processed," making it easier to isolate the cause when something goes wrong.

The transformation pattern differs by layer.

Layer	Transformation Pattern	Reason
Bronze	One-to-one only	Faithfully ingests the source. No transformation.
Silver	Funnel (many-to-one)	Joins and cleanses Bronze tables.
Gold	Funnel (many-to-one)	Aggregates Silver tables with business logic.

The funnel here refers to a transformation pattern that aggregates multiple tables into a single table.

bronze.orders   ─→  silver.orders   ─╮
                                      ├─→ gold.order_summary (funnel)
bronze.contacts ─→  silver.contacts ─╯

Note that fan-out (a pattern that creates multiple tables from a single source) can be expressed by independently defining multiple funnels that reference the same source. There is no need to have it as a dedicated pattern.

---

Prerequisite 2: Technology Stack

This article assumes the following technology stack.

Technology	Purpose
Databricks	Execution environment. Provides Spark cluster and Delta Lake.
Databricks Jobs	Handles scheduled execution, parallel execution, and dependency management for pipelines.
Delta Lake	Table format for writing. Supports updates, deletes, and upserts.
PySpark	Data processing. Spark DataFrames use lazy evaluation, enabling distributed processing of large datasets.
Python	Language for implementing pipeline logic.
YAML	Configuration file format for pipelines.
GitHub Actions / Databricks Asset Bundle	Registers and updates Jobs from CI/CD.

Using Databricks Jobs allows you to declaratively define dependencies between tables and run them in parallel. If one task fails, it doesn't affect others, and retries can be applied individually.

The Solution: Config-Driven Pipelines

The idea behind config-driven design is simple: "What differs per table is only the configuration. The processing code is shared as a single common implementation."

For example, to ingest the orders table from PostgreSQL, all you write is this:

tables:
  - name: raw_orders
    source_type: postgres
    query: "SELECT * FROM orders WHERE updated_at > '{last_run}'"
    destination: bronze.orders
    schedule: "0 * * * *"    # every hour at minute 0
    load_mode: append
    quality:
      on_failure: error
      checks:
        - type: not_null
          columns: [order_id]

To add a new table, you just add one entry to the YAML. You don't touch Python code. Change source_type and it can handle Salesforce or S3. Use load_mode to switch between full overwrite, incremental update, or append. Use schedule to configure execution timing per table individually, and tables with the same schedule are automatically grouped into the same Job. Quality check configuration is also written in the same file and executed automatically.

When you want to fix quality check logic, modifying just one file in the common module reflects the change across all tables. Details are covered in later sections.

---

Overall Structure

Here is an overview of the processing flow for each layer. The Bronze layer simply stores raw data as-is, so no transformation occurs. The Silver/Gold layers add join and transformation processing that references multiple Bronze or Silver tables.

Bronze Layer

Source (PostgreSQL, Salesforce, S3, etc.)
     │ Extract   ← fetch data from source
     ▼
 Raw data
     │ Quality Check   ← data quality check
     ▼
 Quality-verified data
     │ Load   ← write to Delta table
     ▼
 bronze.table (Delta Table)

Silver/Gold Layer — Transform is added on top of the Bronze layer processing.

bronze/silver.table_A  bronze/silver.table_B  …
           │                        │
           └───────────┬────────────┘
                       │ Extract   ← fetch from multiple sources
                       ▼
               Multiple DataFrames
                       │ Transform   ← join, cleanse, aggregate
                       ▼
               Transformed DataFrame
                       │ Quality Check
                       ▼
               Quality-verified data
                       │ Load
                       ▼
               silver/gold.table (Delta Table)

The implementations of Extract, Quality Check, and Load are shared as common modules (sources/ / quality.py / loader.py) across both layers. "Which source to fetch from," "what checks to run," and "which pattern to write with" are all written in YAML configuration, and notebooks read the config to switch behavior. Behavior is controlled purely by configuration, without changing code.

Since each role has its own independent directory, it's immediately clear where to make changes.

pipeline/
├── config/
│   ├── models.py                 # definition of data objects for reading YAML
│   ├── bronze_config.yaml        # Bronze layer: declares tables to ingest
│   └── silver_gold_config.yaml   # Silver/Gold layer: declares tables to transform
├── sources/
│   ├── base.py                   # interface definition for "fetching data"
│   ├── postgres.py               # implementation for fetching from PostgreSQL
│   ├── salesforce.py             # implementation for fetching from Salesforce
│   ├── s3.py                     # implementation for fetching from S3
│   ├── delta.py                  # implementation for fetching from DeltaTable
│   └── registry.py               # mapping of which class to use for each source type
├── transforms/
│   ├── orders.py                 # transformation logic for the orders table
│   └── order_summary.py          # transformation logic for the order_summary table
├── quality.py                    # data quality checks (common)
├── loader.py                     # writing to Delta tables (common)
├── job_deployer.py               # Job registration logic for Databricks Jobs (common)
└── deploy.py                     # entry point executed from CI/CD

---

Common Modules

These are modules shared across both the Bronze and Silver/Gold layers. Understanding them here will make it easier to follow the per-layer implementations later.

config/models.py — Configuration Data Objects

If you pass YAML-loaded results as raw dict through the code, typos and missing settings go unnoticed until runtime, and it becomes hard to track which keys each function needs. By defining them as data objects using dataclass, IDE autocompletion works and the intent of the code becomes clear.

from dataclasses import dataclass, field

@dataclass
class QualityCheck:
    type: str
    columns: list[str] = field(default_factory=list)
    threshold: int = 0
    column: str = ""
    min: float | None = None

@dataclass
class QualityConfig:
    on_failure: str = "error"
    checks: list[QualityCheck] = field(default_factory=list)

@dataclass
class SourceConfig:
    type: str
    alias: str
    url: str = ""
    table: str = ""
    query: str = ""
    path: str = ""

@dataclass
class TableConfig:
    name: str
    destination: str
    load_mode: str = "overwrite"
    quality: QualityConfig = field(default_factory=QualityConfig)
    # for Bronze layer
    source_type: str = ""
    schedule: str = ""          # cron format. e.g.: "0 * * * *" (hourly), "0 6 * * *" (daily at 6am)
    # for Silver/Gold layer
    sources: list[SourceConfig] = field(default_factory=list)
    transform: str = ""
    depends_on: list[str] = field(default_factory=list)
    merge_keys: list[str] = field(default_factory=list)

def load_config(path: str) -> list[TableConfig]:
    import yaml
    from dacite import from_dict
    with open(path) as f:
        raw = yaml.safe_load(f)
    return [from_dict(TableConfig, t) for t in raw["tables"]]

dacite is a library that converts nested dictionaries into dataclass instances. If you want to strengthen validation, the advanced section introduces how to migrate to pydantic.

sources/ — Fetching Data

The fetch logic for each data source is extracted into classes. ExtractorProtocol is an interface that defines the methods all Extractors must implement. Using Python's Protocol eliminates the need for inheritance, so existing classes can be plugged in as-is.

sources/base.py

from typing import Protocol
from pyspark.sql import DataFrame
from config.models import SourceConfig

class ExtractorProtocol(Protocol):
    def extract(self, config: SourceConfig) -> DataFrame:
        ...

sources/postgres.py

from pyspark.sql import DataFrame
from config.models import SourceConfig

class PostgresExtractor:
    def extract(self, config: SourceConfig) -> DataFrame:
        return (
            spark.read.format("jdbc")
            .option("url", config.url)
            .option("query", config.query)
            .load()
        )

The mapping of "which class to use for which source type" is centralized in registry.py. To add a new data source, just add an Extractor class under sources/ and register it with one line in registry.py. No need to modify notebooks or other modules.

For example, to add BigQuery, it would look like this:

sources/registry.py

from sources.base import ExtractorProtocol
from sources.postgres import PostgresExtractor
from sources.salesforce import SalesforceExtractor
from sources.s3 import S3Extractor
from sources.delta import DeltaExtractor
# from sources.bigquery import BigQueryExtractor  ← just add this

EXTRACTORS: dict[str, ExtractorProtocol] = {
    "postgres": PostgresExtractor(),
    "salesforce": SalesforceExtractor(),
    "s3": S3Extractor(),
    "delta": DeltaExtractor(),
    # "bigquery": BigQueryExtractor(),  ← just add this
}

This style is known as the Strategy pattern. Instead of writing branching logic with if/elif chains, processing is encapsulated in classes and switched via a dictionary — a readable and extensible approach. To add a new pattern, just write one class and register it in EXTRACTORS with one line, without touching existing code.

The Silver/Gold layers frequently read from Bronze Delta Tables, so here is an implementation example for DeltaExtractor as well.

sources/delta.py

from pyspark.sql import DataFrame
from config.models import SourceConfig

class DeltaExtractor:
    def extract(self, config: SourceConfig) -> DataFrame:
        return spark.read.format("delta").table(config.table)

quality.py — Data Quality Checks

Executes the checks declared in the configuration file. on_failure: error stops the pipeline when a check fails; on_failure: warn outputs a warning log and continues.

Check types are managed in a dictionary (CHECKERS). Like sources/ and registry.py, it uses the Strategy pattern — to add a new check, just write one class and register it in CHECKERS with one line, without touching existing code.

import logging
from typing import Protocol, TypedDict
from pyspark.sql import DataFrame
from config.models import TableConfig, QualityCheck

logger = logging.getLogger(__name__)

class CheckResult(TypedDict):
    passed: bool
    message: str

class CheckerProtocol(Protocol):
    def check(self, df: DataFrame, rule: QualityCheck) -> CheckResult:
        ...

class NotNullChecker:
    def check(self, df: DataFrame, rule: QualityCheck) -> CheckResult:
        for col in rule.columns:
            null_count: int = df.filter(df[col].isNull()).count()
            if null_count > 0:
                return {"passed": False, "message": f"{col} has {null_count} NULL value(s)"}
        return {"passed": True, "message": ""}

class MinRowsChecker:
    def check(self, df: DataFrame, rule: QualityCheck) -> CheckResult:
        row_count: int = df.count()
        if row_count < rule.threshold:
            return {"passed": False, "message": f"Row count is insufficient ({row_count} < {rule.threshold})"}
        return {"passed": True, "message": ""}

class ValueRangeChecker:
    def check(self, df: DataFrame, rule: QualityCheck) -> CheckResult:
        if rule.min is not None:
            violation: int = df.filter(df[rule.column] < rule.min).count()
            if violation > 0:
                return {"passed": False, "message": f"{rule.column} has {violation} out-of-range value(s)"}
        return {"passed": True, "message": ""}

CHECKERS: dict[str, CheckerProtocol] = {
    "not_null": NotNullChecker(),
    "min_rows": MinRowsChecker(),
    "value_range": ValueRangeChecker(),
}

def run_checks(df: DataFrame, config: TableConfig) -> None:
    on_failure: str = config.quality.on_failure

    for rule in config.quality.checks:
        checker = CHECKERS.get(rule.type)
        if checker is None:
            raise ValueError(f"Unsupported check: {rule.type}")
        result: CheckResult = checker.check(df, rule)
        if not result["passed"]:
            message: str = f"[{config.name}] Quality check failed: {result['message']}"
            if on_failure == "error":
                raise ValueError(message)
            else:
                logger.warning(message)

loader.py — Writing to Delta Tables

A module that centralizes write processing. Multiple write patterns — full overwrite, incremental update, append, and partition-level overwrite — are implemented using the Strategy pattern.

from typing import Protocol
from pyspark.sql import DataFrame
from delta.tables import DeltaTable
from config.models import TableConfig

class LoaderProtocol(Protocol):
    def load(self, df: DataFrame, config: TableConfig) -> None:
        ...

class OverwriteLoader:
    def load(self, df: DataFrame, config: TableConfig) -> None:
        df.write.format("delta").mode("overwrite").saveAsTable(config.destination)

class AppendLoader:
    def load(self, df: DataFrame, config: TableConfig) -> None:
        df.write.format("delta").mode("append").saveAsTable(config.destination)

class MergeLoader:
    def load(self, df: DataFrame, config: TableConfig) -> None:
        merge_condition = " AND ".join([f"target.{k} = source.{k}" for k in config.merge_keys])
        delta_table = DeltaTable.forName(spark, config.destination)
        (
            delta_table.alias("target")
            .merge(df.alias("source"), merge_condition)
            .whenMatchedUpdateAll()
            .whenNotMatchedInsertAll()
            .execute()
        )

class OverwritePartitionLoader:
    def load(self, df: DataFrame, config: TableConfig) -> None:
        (
            df.write.format("delta")
            .mode("overwrite")
            .option("replaceWhere", f"{config.partition_column} = '{config.partition_value}'")
            .saveAsTable(config.destination)
        )

LOADERS: dict[str, LoaderProtocol] = {
    "overwrite": OverwriteLoader(),
    "append": AppendLoader(),
    "merge": MergeLoader(),
    "overwrite_partition": OverwritePartitionLoader(),
}

def load(df: DataFrame, config: TableConfig) -> None:
    loader = LOADERS.get(config.load_mode)
    if loader is None:
        raise ValueError(f"Unsupported load_mode: {config.load_mode}")
    loader.load(df, config)

Four patterns can be specified for load_mode. Add loader classes as needed.

load_mode	Behavior	Use Case
overwrite	Full overwrite	Master data, small tables
merge	Update on key match, insert if not found (Upsert)	Transaction data, data with changes
append	Appends to existing data only	Logs, event data
overwrite_partition	Overwrites only the specified partition	Daily batch where only a specific date needs to be updated

---

Bronze Layer: One-to-One Pipeline

Faithfully ingests the source with no transformation. Specify which Extractor to use with source_type and write directly to the Delta table.

config/bronze_config.yaml

tables:
  - name: raw_orders
    source_type: postgres
    query: "SELECT * FROM orders WHERE updated_at > '{last_run}'"
    destination: bronze.orders
    schedule: "0 * * * *"       # every hour at minute 0
    load_mode: append            # append new records
    quality:
      on_failure: error          # stop pipeline on check failure
      checks:
        - type: not_null
          columns: [order_id]
        - type: min_rows
          threshold: 1

  - name: raw_contacts
    source_type: salesforce
    source_object: Contact
    destination: bronze.contacts
    schedule: "0 * * * *"       # every hour at minute 0 (grouped into the same Job as raw_orders)
    load_mode: overwrite         # full overwrite every time
    quality:
      on_failure: warn           # on check failure, only output a warning log and continue
      checks:
        - type: not_null
          columns: [Id]

  - name: raw_events
    source_type: s3
    path: s3://my-bucket/events/
    destination: bronze.events
    schedule: "0 6 * * *"       # daily at 6am (becomes a separate Job)
    load_mode: append
    quality:
      on_failure: warn
      checks:
        - type: not_null
          columns: [event_id]

bronze_task Notebook

Each Job task registered by deploy.py calls a notebook. Since each table runs as an independent task in parallel, one failure doesn't affect others. It receives table_name as a parameter, loads the configuration, and executes Extract → Quality Check → Load.

from pyspark.sql import DataFrame
from sources.registry import EXTRACTORS
from quality import run_checks
from loader import load
from config.models import TableConfig, load_config

# Receive the table name passed as a parameter from Databricks Jobs
table_name: str = dbutils.widgets.get("table_name")
table: TableConfig = next(t for t in load_config("config/bronze_config.yaml") if t.name == table_name)

# Extract: retrieve the Extractor corresponding to source_type and execute
df: DataFrame = EXTRACTORS[table.source_type].extract(table)

# Quality Check
run_checks(df, table)

# Load
load(df, table)

---

Silver/Gold Layer: Funnel Pipeline

References multiple tables for joining and transformation. All transformation patterns are expressed as funnels (many-to-one). Even when there is only one source, writing it as a funnel means you won't need to change the transform function signature if you add more sources later.

config/silver_gold_config.yaml

List the dependent table names in depends_on. Tables with no dependencies are run in parallel by Databricks Jobs; tables with dependencies run after their dependencies complete.

tables:
  - name: silver_orders
    depends_on: []                        # no dependencies → runs immediately after Bronze completes
    sources:
      - type: delta
        table: bronze.orders
        alias: orders                     # referenced as dfs["orders"] inside the transform
    destination: silver.orders
    transform: transforms.orders.clean
    load_mode: merge
    merge_keys: [order_id]
    quality:
      on_failure: error
      checks:
        - type: not_null
          columns: [order_id, amount]
        - type: min_rows
          threshold: 1000
        - type: value_range
          column: amount
          min: 0

  - name: silver_contacts
    depends_on: []                        # no dependencies → runs concurrently with silver_orders
    sources:
      - type: delta
        table: bronze.contacts
        alias: contacts
    destination: silver.contacts
    transform: transforms.contacts.clean
    load_mode: overwrite
    quality:
      on_failure: warn
      checks:
        - type: not_null
          columns: [contact_id]

  - name: gold_order_summary
    depends_on: [silver_orders, silver_contacts]  # runs after both complete
    sources:
      - type: delta
        table: silver.orders
        alias: orders
      - type: delta
        table: silver.contacts
        alias: contacts
    destination: gold.order_summary
    transform: transforms.order_summary.build
    load_mode: overwrite
    quality:
      on_failure: error
      checks:
        - type: not_null
          columns: [order_id, contact_id]

transforms/ — Transformation Logic

All transform functions receive dfs: dict[str, DataFrame] and return a single DataFrame. Each key in dfs corresponds to the alias in the configuration file. Even when there is only one source, writing with the same signature means the transform function definition won't need to change if more sources are added later.

transforms/orders.py — Funnel with a single source

from pyspark.sql import DataFrame
from pyspark.sql import functions as F
from pyspark.sql.window import Window

def clean(dfs: dict[str, DataFrame]) -> DataFrame:
    df = dfs["orders"]   # retrieve using the alias key
    return (
        df
        .drop("_internal_id", "_raw_payload")
        .withColumn("amount", F.col("amount").cast("double"))
        .withColumn("ordered_at", F.to_timestamp("ordered_at", "yyyy-MM-dd HH:mm:ss"))
        .filter(F.col("order_id").isNotNull())
        .filter(F.col("amount") > 0)
        .withColumn("status", F.lower(F.trim(F.col("status"))))
        .withColumn("order_year", F.year("ordered_at"))
        .withColumn("order_month", F.month("ordered_at"))
        .withColumn(
            "row_num",
            F.row_number().over(
                Window.partitionBy("order_id").orderBy(F.desc("ordered_at"))
            )
        )
        .filter(F.col("row_num") == 1)
        .drop("row_num")
    )

transforms/order_summary.py — Funnel with multiple sources

from pyspark.sql import DataFrame

def build(dfs: dict[str, DataFrame]) -> DataFrame:
    orders = dfs["orders"]
    contacts = dfs["contacts"]
    return (
        orders
        .join(contacts, orders["customer_id"] == contacts["Id"], "left")
        .select("order_id", "contact_id", "amount", "ordered_at")
    )

silver_gold_task Notebook

Receives table_name as a parameter, loads the configuration, and executes Extract → Transform → Quality Check → Load. silver_orders and silver_contacts run in parallel, and gold_order_summary runs after both complete.

silver_orders   ─╮
                  ├─→ gold_order_summary
silver_contacts ─╯

import importlib
from types import ModuleType
from typing import Callable
from pyspark.sql import DataFrame
from sources.registry import EXTRACTORS
from quality import run_checks
from loader import load
from config.models import TableConfig, load_config

# Receive the table name passed as a parameter from Databricks Jobs
table_name: str = dbutils.widgets.get("table_name")
table: TableConfig = next(t for t in load_config("config/silver_gold_config.yaml") if t.name == table_name)

# Extract: retrieve multiple sources as a dictionary keyed by alias
# Spark DataFrames use lazy evaluation, so data is not actually read at this point.
# Data is actually processed when an action is executed inside the transform.
dfs: dict[str, DataFrame] = {source.alias: EXTRACTORS[source.type].extract(source.table) for source in table.sources}

# Transform: parse the string "transforms.orders.clean" from YAML into module and function name and dynamically load
# → calls the clean() function in transforms/orders.py at runtime
module_path, func_name = table.transform.rsplit(".", 1)
module: ModuleType = importlib.import_module(module_path)
transform_fn: Callable[[dict[str, DataFrame]], DataFrame] = getattr(module, func_name)
df: DataFrame = transform_fn(dfs)

# Quality Check
run_checks(df, table)

# Load
load(df, table)

---

Deployment: Registering with Databricks Jobs

Job registration is done from the CI/CD pipeline (GitHub Actions, Databricks Asset Bundle, etc.). It is re-run when the configuration file changes (e.g., tables added or removed) to update the Jobs.

Registration logic is consolidated in job_deployer.py. Bronze tables are grouped by schedule from bronze_config.yaml, and one Job is created per schedule. Silver/Gold tasks that depend on the corresponding Bronze are automatically added as downstream tasks to each Job. Since all tables share the same notebook and switch the table to process via the table_name parameter, even with 100 tables you only need 2 notebooks.

After registration, the Job task graph looks like this (assuming raw_orders and raw_contacts run hourly, and raw_events runs daily):

pipeline_0_*_*_*_* (hourly)
  ├── [Bronze] raw_orders
  ├── [Bronze] raw_contacts
  ├── [Silver] silver_orders    depends_on: [raw_orders]
  └── [Silver] silver_contacts  depends_on: [raw_contacts]

pipeline_0_6_*_*_* (daily)
  ├── [Bronze] raw_events
  ├── [Silver] silver_events        depends_on: [raw_events]
  └── [Gold]   gold_order_summary   depends_on: [silver_orders, silver_contacts, silver_events, raw_events]
                                     ↑ raw_events in depends_on assigns this to this Job
                                     ↑ silver_orders and silver_contacts are assumed to have already completed in the hourly Job

How to write depends_on in silver_gold_config.yaml

depends_on can include both Silver/Gold table names (for task ordering) and Bronze table names (for Job group assignment). An entry with a Bronze name means "put me in the Job that runs the schedule for that Bronze."

# When depending only on Bronze with the same schedule
- name: silver_orders
  depends_on: [raw_orders]        # Bronze name → goes into hourly Job + task ordering

# When waiting for Bronze with a different schedule
- name: gold_order_summary
  depends_on: [silver_orders, silver_contacts, silver_events, raw_events]
  #            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^  S/G inter-dependencies (task ordering)
  #                                                                 ^^^^^^^^^^
  #                                                                 Bronze name → goes into daily Job

job_deployer.py

from databricks.sdk import WorkspaceClient
from databricks.sdk.service.jobs import (
    TaskDependency,
    NotebookTask,
    Task,
    CronSchedule,
)
from config.models import TableConfig

def _bronze_names(bronze_tables: list[TableConfig]) -> set[str]:
    """Returns a set of Bronze task names. e.g.: {"raw_orders", "raw_contacts"}"""
    return {t.name for t in bronze_tables}

def _sg_tables_for_group(
    sg_tables: list[TableConfig],
    group_bronze_names: set[str],
    all_bronze_names: set[str],
) -> list[TableConfig]:
    """Returns the Silver/Gold tables that should be included in this Bronze group.

    Starting from those whose depends_on directly contains a Bronze name in this group,
    performs BFS over S/G inter-dependencies to track transitive dependencies as well.
    """
    # Start with those that directly contain a Bronze name in this group in depends_on
    directly_dependent: set[str] = {
        t.name
        for t in sg_tables
        if group_bronze_names & set(t.depends_on)
    }

    # Transitively dependent (BFS)
    # Only traverse S/G inter-dependencies (Bronze names are excluded as they are used for group assignment)
    included: set[str] = set(directly_dependent)
    frontier = list(directly_dependent)
    while frontier:
        current = frontier.pop()
        for t in sg_tables:
            if t.name in included:
                continue
            sg_deps = [d for d in t.depends_on if d not in all_bronze_names]
            if current in sg_deps:
                included.add(t.name)
                frontier.append(t.name)

    return [t for t in sg_tables if t.name in included]

def _build_bronze_tasks(
    bronze_tables: list[TableConfig],
    notebook_path: str,
) -> list[Task]:
    return [
        Task(
            task_key=t.name,
            depends_on=[],
            notebook_task=NotebookTask(
                notebook_path=notebook_path,
                base_parameters={"table_name": t.name},
            ),
        )
        for t in bronze_tables
    ]

def _build_sg_tasks(
    sg_tables: list[TableConfig],
    notebook_path: str,
) -> list[Task]:
    """Builds Silver/Gold tasks.

    Values from depends_on in YAML are converted directly to TaskDependency.
    Both Bronze names and S/G names are treated as task dependencies,
    and Databricks Jobs controls the order according to the task graph.
    """
    return [
        Task(
            task_key=t.name,
            depends_on=[TaskDependency(task_key=dep) for dep in t.depends_on],
            notebook_task=NotebookTask(
                notebook_path=notebook_path,
                base_parameters={"table_name": t.name},
            ),
        )
        for t in sg_tables
    ]

def deploy_jobs_by_schedule(
    bronze_tables: list[TableConfig],
    sg_tables: list[TableConfig],
    bronze_notebook_path: str,
    sg_notebook_path: str,
) -> None:
    """Groups Bronze by schedule and registers the corresponding Silver/Gold
    as downstream tasks in the same Job.
    """
    all_bronze_names = _bronze_names(bronze_tables)

    # Group by schedule
    schedule_groups: dict[str, list[TableConfig]] = {}
    for t in bronze_tables:
        schedule_groups.setdefault(t.schedule, []).append(t)

    client = WorkspaceClient()

    for schedule, group_bronze in schedule_groups.items():
        group_bronze_names = _bronze_names(group_bronze)
        group_sg = _sg_tables_for_group(sg_tables, group_bronze_names, all_bronze_names)

        tasks: list[Task] = (
            _build_bronze_tasks(group_bronze, bronze_notebook_path)
            + _build_sg_tasks(group_sg, sg_notebook_path)
        )

        job_name = f"pipeline_{schedule.replace(' ', '_')}"

        client.jobs.create(
            name=job_name,
            tasks=tasks,
            schedule=CronSchedule(
                quartz_cron_expression=schedule,
                timezone_id="Asia/Tokyo",
            ),
        )

deploy.py (executed from CI/CD)

from job_deployer import deploy_jobs_by_schedule
from config.models import load_config

deploy_jobs_by_schedule(
    bronze_tables=load_config("config/bronze_config.yaml"),
    sg_tables=load_config("config/silver_gold_config.yaml"),
    bronze_notebook_path="/notebooks/bronze_task",
    sg_notebook_path="/notebooks/silver_gold_task",
)

---

Summary

Introducing config-driven pipelines resolves the problems raised at the start as follows.

The problem of similar code appearing everywhere Extract, Quality, and Load processing are implemented once in common modules. Even if the connection URL changes, only one file needs to be modified.

The problem of having to write a Python file every time a new table is added Just add one entry to the YAML. Review only needs to look at the configuration file, not Python code.

The problem of quality checks and error handling being implemented differently across each pipeline quality.py and loader.py are consolidated as common modules. Because they use the Strategy pattern, adding new checks or write patterns doesn't require touching existing code.

Here is a summary of where to make each type of change.

What you want to do	Where to change
Add a Bronze table	Just add one entry to bronze_config.yaml
Add a Silver/Gold table	Just add one entry to silver_gold_config.yaml
Join multiple sources	Add entries to the sources list and retrieve with dfs["alias"] in the transform
Add a new source type	Add an Extractor to sources/ and register it in registry.py
Change transformation logic	Modify only the relevant file in transforms/
Change the load pattern	Just switch load_mode in the config
Add or change a quality check	Just edit checks in the config
Change behavior on check failure	Switch on_failure in the config between error / warn
Define dependencies between tables	Just edit depends_on in silver_gold_config.yaml
Change Bronze schedule	Just change schedule in bronze_config.yaml. Bronze with the same value is automatically grouped into the same Job
Run S/G after Bronze with a different schedule	Just add the Bronze table name to depends_on in silver_gold_config.yaml
Parallelize	Execute deploy.py from CI/CD to register the task graph with Databricks Jobs

---

Advanced: Making It More Robust

Once the core implementation is solid, adding the following two items can increase the reliability of the pipeline.

Validation with pydantic

In this article, configuration objects are defined with dataclass, but migrating to pydantic causes validation to run at YAML load time, detecting configuration errors before the pipeline crashes midway.

from pydantic import BaseModel, field_validator
from typing import Literal

class QualityConfig(BaseModel):
    on_failure: Literal["error", "warn"] = "error"  # ← anything other than error/warn is a validation error
    checks: list[QualityCheck] = []

class TableConfig(BaseModel):
    name: str
    destination: str
    load_mode: Literal["overwrite", "append", "merge", "overwrite_partition"] = "overwrite"  # ← typos caught here
    ...

    @field_validator("merge_keys")
    @classmethod
    def merge_keys_required_for_merge(cls, v, info):
        if info.data.get("load_mode") == "merge" and not v:
            raise ValueError("merge_keys must be specified when load_mode=merge")
        return v

def load_config(path: str) -> list[TableConfig]:
    import yaml
    with open(path) as f:
        raw = yaml.safe_load(f)
    return [TableConfig(**t) for t in raw["tables"]]  # ← validation runs here

By enumerating allowed values with Literal, typos like load_mode: mereg or invalid values like on_failure: stop are caught as errors at YAML load time. dacite is no longer needed, and pydantic handles both conversion and validation on its own.

Example of Adding a Check

Leveraging the extensibility of the Strategy pattern, you can add custom checks. For example, to add a NoDuplicatesChecker that checks "are there any duplicate rows?":

# 1. Add the class
class NoDuplicatesChecker:
    def check(self, df: DataFrame, rule: QualityCheck) -> CheckResult:
        total: int = df.count()
        distinct: int = df.select(rule.column).distinct().count()
        if total != distinct:
            return {"passed": False, "message": f"{rule.column} has {total - distinct} duplicate(s)"}
        return {"passed": True, "message": ""}

# 2. Add one line to CHECKERS
CHECKERS: dict[str, CheckerProtocol] = {
    "not_null": NotNullChecker(),
    "min_rows": MinRowsChecker(),
    "value_range": ValueRangeChecker(),
    "no_duplicates": NoDuplicatesChecker(),  # ← just add this
}

After adding, you can use it simply by writing type: no_duplicates in the checks section of your YAML.