pipeline-stage

Name: pipeline-stage
Author: zakelfassi

17stars0forks

How to install pipeline-stage?

zakelfassi/skills-driven-development/pipeline-stage

$npx -y skills add zakelfassi/skills-driven-development --skill pipeline-stage

Installs into the current project.

Summary

Scaffold a new transform stage in the data pipeline — create the transformation script, schema contract, idempotency logic, and tests. Use when adding a new dbt model or pandas transform, when a new business metric needs a dedicated stage, or when asked to "add a {name} stage to

Files · 1

View on GitHub

SKILL.md

1	# Pipeline Stage
2
3	Create a new transform stage with idempotency guarantees, a schema contract, and tests.
4
5	## Inputs
6	- Stage name (snake_case, e.g., `customer_ltv`)
7	- Input tables/models (list of upstream stage names or raw tables)
8	- Output table name (usually matches stage name)
9	- Grain (the primary key or unique key, e.g., `customer_id`, `(order_id, date)`)
10	- Layer (`staging`, `intermediate`, `marts`)
11
12	## Steps
13
14	1. Create the transform file
15
16	For dbt:
17	```
18	models/{layer}/{stage_name}.sql
19	```
20	```sql
21	{{ config(
22	materialized='table',
23	unique_key='{grain}'
24	) }}
25
26	select
27	{grain},
28	-- TODO: add business logic
29	current_timestamp as updated_at
30	from {{ ref('{input_table}') }}
31	```
32
33	For pandas ETL:
34	```
35	pipelines/transforms/{stage_name}/transform.py
36	```
37	```python
38	def run(df: pd.DataFrame) -> pd.DataFrame:
39	"""Transform {input_table} → {stage_name}."""
40	# TODO: add business logic
41	return df
42	```
43
44	2. Define the schema contract
45	Create `models/{layer}/schema/{stage_name}.yaml` (dbt) or `pipelines/transforms/{stage_name}/schema.py`:
46	```yaml
47	- name: {stage_name}
48	columns:
49	- name: {grain}
50	tests:
51	- unique
52	- not_null
53	```
54	Every non-nullable column must have `not_null` test; every unique key must have `unique` test.
55
56	3. Add idempotency logic
57	- For `materialized='table'`: dbt handles full replacement — no extra work.
58	- For incremental models: use `is_incremental()` filter on `updated_at` or an event timestamp.
59	- For pandas: the output must be deterministic given the same input; add a dedup step on `{grain}`.
60
61	4. Write tests
62	```
63	tests/transforms/test_{stage_name}.py
64	```
65	Required tests:
66	- Input fixture → expected output shape (column names, row count)
67	- Idempotency: running twice produces identical output
68	- Null check: no nulls in required columns after transform
69
70	5. Register in the pipeline DAG
71	Add the stage after its upstream dependencies:
72	```python
73	# dags/pipeline.py
74	{stage_name}_task = DbtRunOperator(
75	task_id="{stage_name}",
76	models="{stage_name}",
77	)
78	{upstream_task} >> {stage_name}_task
79	```
80
81	6. Run locally
82	```bash
83	dbt run --select {stage_name}
84	dbt test --select {stage_name}
85	# or for pandas:
86	python -m pytest tests/transforms/test_{stage_name}.py -v
87	```
88
89	## Conventions
90	- Layer hierarchy: `raw` → `staging` → `intermediate` → `marts`
91	- Never skip a layer (e.g., don't read from `raw` in a `marts` model)
92	- All stages have at least one `unique` + `not_null` test on the grain column
93	- Incremental models use `updated_at` as the watermark; add it to every model
94
95	## Edge Cases
96	- Fan-out (multiple downstream consumers): Create the stage at the `intermediate` layer; let downstream `marts` models reference it.
97	- Slowly changing dimension (SCD): Use dbt's `snapshot` materialization or add `valid_from`/`valid_to` columns manually.
98	- Cross-database join: Materialize both inputs to the same database first, then join; cross-database SQL is not portable.
99	- Very wide table (>200 columns): Split into a core model plus an extension model; document the split in the schema YAML.

People who liked this also viewed

6 picks

Type

pipeline-stage

17stars0forks

View on GitHub

How to install pipeline-stage?

zakelfassi/skills-driven-development/pipeline-stage

$npx -y skills add zakelfassi/skills-driven-development --skill pipeline-stage

Installs into the current project.

Summary

Files · 1

View on GitHub

SKILL.md

1	# Pipeline Stage
2
3	Create a new transform stage with idempotency guarantees, a schema contract, and tests.
4
5	## Inputs
6	- Stage name (snake_case, e.g., `customer_ltv`)
7	- Input tables/models (list of upstream stage names or raw tables)
8	- Output table name (usually matches stage name)
9	- Grain (the primary key or unique key, e.g., `customer_id`, `(order_id, date)`)
10	- Layer (`staging`, `intermediate`, `marts`)
11
12	## Steps
13
14	1. Create the transform file
15
16	For dbt:
17	```
18	models/{layer}/{stage_name}.sql
19	```
20	```sql
21	{{ config(
22	materialized='table',
23	unique_key='{grain}'
24	) }}
25
26	select
27	{grain},
28	-- TODO: add business logic
29	current_timestamp as updated_at
30	from {{ ref('{input_table}') }}
31	```
32
33	For pandas ETL:
34	```
35	pipelines/transforms/{stage_name}/transform.py
36	```
37	```python
38	def run(df: pd.DataFrame) -> pd.DataFrame:
39	"""Transform {input_table} → {stage_name}."""
40	# TODO: add business logic
41	return df
42	```
43
44	2. Define the schema contract
45	Create `models/{layer}/schema/{stage_name}.yaml` (dbt) or `pipelines/transforms/{stage_name}/schema.py`:
46	```yaml
47	- name: {stage_name}
48	columns:
49	- name: {grain}
50	tests:
51	- unique
52	- not_null
53	```
54	Every non-nullable column must have `not_null` test; every unique key must have `unique` test.
55
56	3. Add idempotency logic
57	- For `materialized='table'`: dbt handles full replacement — no extra work.
58	- For incremental models: use `is_incremental()` filter on `updated_at` or an event timestamp.
59	- For pandas: the output must be deterministic given the same input; add a dedup step on `{grain}`.
60
61	4. Write tests
62	```
63	tests/transforms/test_{stage_name}.py
64	```
65	Required tests:
66	- Input fixture → expected output shape (column names, row count)
67	- Idempotency: running twice produces identical output
68	- Null check: no nulls in required columns after transform
69
70	5. Register in the pipeline DAG
71	Add the stage after its upstream dependencies:
72	```python
73	# dags/pipeline.py
74	{stage_name}_task = DbtRunOperator(
75	task_id="{stage_name}",
76	models="{stage_name}",
77	)
78	{upstream_task} >> {stage_name}_task
79	```
80
81	6. Run locally
82	```bash
83	dbt run --select {stage_name}
84	dbt test --select {stage_name}
85	# or for pandas:
86	python -m pytest tests/transforms/test_{stage_name}.py -v
87	```
88
89	## Conventions
90	- Layer hierarchy: `raw` → `staging` → `intermediate` → `marts`
91	- Never skip a layer (e.g., don't read from `raw` in a `marts` model)
92	- All stages have at least one `unique` + `not_null` test on the grain column
93	- Incremental models use `updated_at` as the watermark; add it to every model
94
95	## Edge Cases
96	- Fan-out (multiple downstream consumers): Create the stage at the `intermediate` layer; let downstream `marts` models reference it.
97	- Slowly changing dimension (SCD): Use dbt's `snapshot` materialization or add `valid_from`/`valid_to` columns manually.
98	- Cross-database join: Materialize both inputs to the same database first, then join; cross-database SQL is not portable.
99	- Very wide table (>200 columns): Split into a core model plus an extension model; document the split in the schema YAML.

People who liked this also viewed

6 picks

Type