Pipeline Stage

Pipeline Stage

Create a new transform stage with idempotency guarantees, a schema contract, and tests.

Inputs

• Stage name (snake_case, e.g., customer_ltv)
• Input tables/models (list of upstream stage names or raw tables)
• Output table name (usually matches stage name)
• Grain (the primary key or unique key, e.g., customer_id, (order_id, date))
• Layer (staging, intermediate, marts)

Steps

1. Create the transform file

For dbt: `` models/{layer}/{stage_name}.sql ` ``sql {{ config( materialized='table', unique_key='{grain}' ) }}

select {grain}, -- TODO: add business logic currenttimestamp as updatedat from {{ ref('{input_table}') }} ```

For pandas ETL: `` pipelines/transforms/{stage_name}/transform.py ` `python def run(df: pd.DataFrame) -> pd.DataFrame: """Transform {input_table} → {stage_name}.""" # TODO: add business logic return df ``

1. Define the schema contract

Create models/{layer}/schema/{stage_name}.yaml (dbt) or pipelines/transforms/{stage_name}/schema.py: ```yaml

• name: {stage_name}

columns:

• name: {grain}

tests:

• unique
• not_null

`` Every non-nullable column must have not_null test; every unique key must have unique` test.

1. Add idempotency logic

• For materialized='table': dbt handles full replacement — no extra work.
• For incremental models: use is_incremental() filter on updated_at or an event timestamp.
• For pandas: the output must be deterministic given the same input; add a dedup step on {grain}.

1. Write tests

`` tests/transforms/test_{stage_name}.py `` Required tests:

• Input fixture → expected output shape (column names, row count)
• Idempotency: running twice produces identical output
• Null check: no nulls in required columns after transform

1. Register in the pipeline DAG

Add the stage after its upstream dependencies: ``python # dags/pipeline.py {stage_name}_task = DbtRunOperator( task_id="{stage_name}", models="{stage_name}", ) {upstream_task} >> {stage_name}_task ``

1. Run locally

``bash dbt run --select {stage_name} dbt test --select {stage_name} # or for pandas: python -m pytest tests/transforms/test_{stage_name}.py -v ``

Conventions

• Layer hierarchy: raw → staging → intermediate → marts
• Never skip a layer (e.g., don't read from raw in a marts model)
• All stages have at least one unique + not_null test on the grain column
• Incremental models use updated_at as the watermark; add it to every model

Edge Cases

• Fan-out (multiple downstream consumers): Create the stage at the intermediate layer; let downstream marts models reference it.
• Slowly changing dimension (SCD): Use dbt's snapshot materialization or add valid_from/valid_to columns manually.
• Cross-database join: Materialize both inputs to the same database first, then join; cross-database SQL is not portable.
• Very wide table (>200 columns): Split into a core model plus an extension model; document the split in the schema YAML.

Source & license

This open-source skill is cataloged on AgentStack and links to its original source — we do not rehost the code.

• Author: zakelfassi
• Source: zakelfassi/skills-driven-development
• License: MIT
• Homepage: https://zakelfassi.github.io/skills-driven-development/

Install and usage instructions live in the source repository linked above.