Data Engineering
Pipeline Architecture
ETL vs ELT
| Pattern |
When to Use |
Tools |
| ETL |
Transform before loading, data quality critical |
Airflow + custom, Spark |
| ELT |
Raw → warehouse → transform in-place |
Fivetran + dbt, Airbyte + dbt |
Orchestration
Apache Airflow:
from airflow.decorators import dag, task
from datetime import datetime
@dag(schedule="@daily", start_date=datetime(2024, 1, 1), catchup=False)
def my_pipeline():
@task()
def extract() -> dict:
return {"data": "extracted"}
@task()
def transform(data: dict) -> dict:
return {"transformed": True}
@task()
def load(data: dict):
# Load to warehouse
pass
raw = extract()
transformed = transform(raw)
load(transformed)
my_pipeline()
Dagster (recommended for new projects):
from dagster import asset, Definitions
@asset
def raw_users():
return extract_from_source()
@asset
def cleaned_users(raw_users):
return clean_and_validate(raw_users)
dbt Transformations
-- models/marts/dim_customers.sql
{{ config(materialized='table', schema='marts') }}
WITH source AS (
SELECT * FROM {{ ref('stg_customers') }}
),
orders AS (
SELECT customer_id, COUNT(*) as order_count, SUM(amount) as total_spent
FROM {{ ref('stg_orders') }}
GROUP BY customer_id
)
SELECT
s.customer_id,
s.name,
s.email,
COALESCE(o.order_count, 0) as lifetime_orders,
COALESCE(o.total_spent, 0) as lifetime_value
FROM source s
LEFT JOIN orders o ON s.customer_id = o.customer_id
Stream Processing
Apache Kafka:
from confluent_kafka import Producer, Consumer
# Producer
producer = Producer({'bootstrap.servers': 'localhost:9092'})
producer.produce('events', key='user_123', value=json.dumps(event))
producer.flush()
# Consumer
consumer = Consumer({
'bootstrap.servers': 'localhost:9092',
'group.id': 'my-group',
'auto.offset.reset': 'earliest'
})
consumer.subscribe(['events'])
Data Warehouse Schema Design
Star Schema
- Fact tables: Measurable events (orders, clicks, transactions)
- Dimension tables: Descriptive context (customers, products, dates)
- Slowly Changing Dimensions: Type 1 (overwrite), Type 2 (versioned rows), Type 3 (previous column)
Data Quality
- Great Expectations: Schema validation, statistical tests, custom expectations
- dbt tests:
not_null, unique, accepted_values, relationships, custom SQL tests
- Data contracts: Schema evolution policies, backward compatibility requirements
Key Patterns
- Idempotent pipelines: Same input always produces same output, safe to rerun
- Incremental models: Process only new/changed data, use
updated_at watermarks
- Dead letter queues: Route failed records for inspection without blocking pipeline
- Backfill strategy: Time-partitioned tables enable targeted historical reprocessing
1---2name: data-engineering3description: ETL/ELT pipelines, data warehousing (BigQuery, Snowflake, Redshift), stream processing (Kafka, Spark Streaming), orchestration (Airflow, Dagster, Prefect), dbt transformations, and data lake architecture. Use when building data pipelines, designing warehouse schemas, or implementing real-time data processing.4---5
6# Data Engineering
7
8## Pipeline Architecture
9
10### ETL vs ELT
11| Pattern | When to Use | Tools |
12|---------|-------------|-------|
13| **ETL** | Transform before loading, data quality critical | Airflow + custom, Spark |
14| **ELT** | Raw → warehouse → transform in-place | Fivetran + dbt, Airbyte + dbt |
15
16### Orchestration
17
18**Apache Airflow:**
19```python
20from airflow.decorators import dag, task
21from datetime import datetime
22
23@dag(schedule="@daily", start_date=datetime(2024, 1, 1), catchup=False)
24def my_pipeline():
25 @task()
26 def extract() -> dict:
27 return {"data": "extracted"}
28
29 @task()
30 def transform(data: dict) -> dict:
31 return {"transformed": True}
32
33 @task()
34 def load(data: dict):
35 # Load to warehouse
36 pass
37
38 raw = extract()
39 transformed = transform(raw)
40 load(transformed)
41
42my_pipeline()
43```
44
45**Dagster (recommended for new projects):**
46```python
47from dagster import asset, Definitions
48
49@asset
50def raw_users():
51 return extract_from_source()
52
53@asset
54def cleaned_users(raw_users):
55 return clean_and_validate(raw_users)
56```
57
58## dbt Transformations
59
60```sql
61-- models/marts/dim_customers.sql
62{{ config(materialized='table', schema='marts') }}
63
64WITH source AS (
65 SELECT * FROM {{ ref('stg_customers') }}
66),
67orders AS (
68 SELECT customer_id, COUNT(*) as order_count, SUM(amount) as total_spent
69 FROM {{ ref('stg_orders') }}
70 GROUP BY customer_id
71)
72SELECT
73 s.customer_id,
74 s.name,
75 s.email,
76 COALESCE(o.order_count, 0) as lifetime_orders,
77 COALESCE(o.total_spent, 0) as lifetime_value
78FROM source s
79LEFT JOIN orders o ON s.customer_id = o.customer_id
80```
81
82## Stream Processing
83
84**Apache Kafka:**
85```python
86from confluent_kafka import Producer, Consumer
87
88# Producer
89producer = Producer({'bootstrap.servers': 'localhost:9092'})
90producer.produce('events', key='user_123', value=json.dumps(event))
91producer.flush()
92
93# Consumer
94consumer = Consumer({
95 'bootstrap.servers': 'localhost:9092',
96 'group.id': 'my-group',
97 'auto.offset.reset': 'earliest'
98})
99consumer.subscribe(['events'])
100```
101
102## Data Warehouse Schema Design
103
104### Star Schema
105- **Fact tables:** Measurable events (orders, clicks, transactions)
106- **Dimension tables:** Descriptive context (customers, products, dates)
107- **Slowly Changing Dimensions:** Type 1 (overwrite), Type 2 (versioned rows), Type 3 (previous column)
108
109### Data Quality
110- **Great Expectations:** Schema validation, statistical tests, custom expectations
111- **dbt tests:** `not_null`, `unique`, `accepted_values`, `relationships`, custom SQL tests
112- **Data contracts:** Schema evolution policies, backward compatibility requirements
113
114## Key Patterns
115- **Idempotent pipelines:** Same input always produces same output, safe to rerun
116- **Incremental models:** Process only new/changed data, use `updated_at` watermarks
117- **Dead letter queues:** Route failed records for inspection without blocking pipeline
118- **Backfill strategy:** Time-partitioned tables enable targeted historical reprocessing
travisjneuman