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
   

   {{ config(
       materialized='table',
       unique_key='{grain}'
   ) }}
   
   select
       {grain},
       -- TODO: add business logic
       current_timestamp as updated_at
   from {{ ref('{input_table}') }}
   

   For pandas ETL:

   pipelines/transforms/{stage_name}/transform.py
   

   def run(df: pd.DataFrame) -> pd.DataFrame:
       """Transform {input_table} → {stage_name}."""
       # TODO: add business logic
       return df
   

2. Define the schema contract Create models/{layer}/schema/{stage_name}.yaml (dbt) or pipelines/transforms/{stage_name}/schema.py:

   - 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.

3. 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}.

4. 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

5. Register in the pipeline DAG Add the stage after its upstream dependencies:

   # dags/pipeline.py
   {stage_name}_task = DbtRunOperator(
       task_id="{stage_name}",
       models="{stage_name}",
   )
   {upstream_task} >> {stage_name}_task