Data Engineer Skill

Data Engineer

Purpose

Provides expert data engineering capabilities for building scalable data pipelines, ETL/ELT workflows, data lakes, and data warehouses. Specializes in distributed data processing, stream processing, data quality, and modern data stack technologies (Airflow, dbt, Spark, Kafka) with focus on reliability and cost optimization.

When to Use

• Designing end-to-end data pipelines from source to consumption layer
• Implementing ETL/ELT workflows with error handling and data quality checks
• Building data lakes or data warehouses with optimal storage and querying
• Setting up real-time stream processing (Kafka, Flink, Kinesis)
• Optimizing data infrastructure costs (storage tiering, compute efficiency)
• Implementing data governance and compliance (GDPR, data lineage)
• Migrating legacy data systems to modern data platforms

Quick Start

Invoke this skill when:

• Designing end-to-end data pipelines from source to consumption layer
• Implementing ETL/ELT workflows with error handling and data quality checks
• Building data lakes or data warehouses with optimal storage and querying
• Setting up real-time stream processing (Kafka, Flink, Kinesis)
• Optimizing data infrastructure costs (storage tiering, compute efficiency)
• Implementing data governance and compliance (GDPR, data lineage)

Do NOT invoke when:

• Only SQL query optimization needed (use database-optimizer instead)
• Machine learning model development (use ml-engineer or data-scientist)
• Simple data analysis or visualization (use data-analyst)
• Database administration tasks (use database-administrator)
• API integration without data transformation (use backend-developer)

Decision Framework

Pipeline Architecture Selection

├─ Batch Processing?
│   ├─ Daily/hourly schedules → Airflow + dbt
│   │   Pros: Mature ecosystem, SQL-based transforms
│   │   Cost: Low-medium
│   │
│   ├─ Large-scale (TB+) → Spark (EMR/Databricks)
│   │   Pros: Distributed processing, handles scale
│   │   Cost: Medium-high (compute-intensive)
│   │
│   └─ Simple transforms → dbt Cloud or Fivetran
│       Pros: Managed, low maintenance
│       Cost: Medium (SaaS pricing)
│
├─ Stream Processing?
│   ├─ Event streaming → Kafka + Flink
│   │   Pros: Low latency, exactly-once semantics
│   │   Cost: High (always-on infrastructure)
│   │
│   ├─ AWS native → Kinesis + Lambda
│   │   Pros: Serverless, auto-scaling
│   │   Cost: Variable (pay per use)
│   │
│   └─ Simple CDC → Debezium + Kafka Connect
│       Pros: Database change capture
│       Cost: Medium
│
└─ Hybrid (Batch + Stream)?
    └─ Lambda Architecture or Kappa Architecture
        Lambda: Separate batch/speed layers
        Kappa: Single stream-first approach

Data Storage Selection

ETL vs ELT Decision

Core Patterns

Pattern 1: Idempotent Partition Overwrite

Use case: Safely re-run batch jobs without creating duplicates.

# PySpark example: Overwrite partition based on execution date
def write_daily_partition(df, target_table, execution_date):
    (df
     .write
     .mode("overwrite")
     .partitionBy("process_date")
     .option("partitionOverwriteMode", "dynamic")
     .format("parquet")
     .saveAsTable(target_table))

Pattern 2: Slowly Changing Dimension Type 2 (SCD2)

Use case: Track history of changes without losing past states.

-- dbt implementation of SCD2
{{ config(materialized='incremental', unique_key='user_id') }}

SELECT 
    user_id, address, email, status, updated_at,
    LEAD(updated_at, 1, '9999-12-31') OVER (
        PARTITION BY user_id ORDER BY updated_at
    ) as valid_to
FROM {{ source('raw', 'users') }}

Pattern 3: Dead Letter Queue (DLQ) for Streaming

Use case: Handle malformed messages without stopping the pipeline.

Pattern 4: Data Quality Circuit Breaker

Use case: Stop pipeline execution if data quality drops below threshold.

Quality Checklist

Data Pipeline

• [ ] Idempotent (safe to retry)
• [ ] Schema validation enforced
• [ ] Error handling with retries
• [ ] Data quality checks automated
• [ ] Monitoring and alerting configured
• [ ] Lineage documented

Performance

• [ ] Pipeline completes within SLA (e.g., <1 hour)
• [ ] Incremental loading where applicable
• [ ] Partitioning strategy optimized
• [ ] Query performance <30 seconds (P95)

Cost Optimization

• [ ] Storage tiering implemented (hot/warm/cold)
• [ ] Compute auto-scaling configured
• [ ] Query cost monitoring active
• [ ] Compression enabled (Parquet/ORC)

Additional Resources

• Detailed Technical Reference: See REFERENCE.md
• Code Examples & Patterns: See EXAMPLES.md

Data Engineer - Code Examples & Patterns

Anti-Pattern: No Idempotency in Data Pipelines

What it looks like (BAD):

def daily_etl():
    # Extract new data
    new_data = extract_from_api()
    
    # Transform
    transformed_data = transform(new_data)
    
    # Load (append mode)
    transformed_data.to_sql('fact_orders', engine, if_exists='append')

# Problem: If pipeline fails halfway and restarts → duplicates!

Why it fails:

• Pipeline failures create duplicate records
• No way to safely retry failed runs
• Data integrity compromised
• Counts and aggregations become incorrect

Correct approach:

import uuid
from datetime import datetime

def daily_etl(run_date=None, run_id=None):
    # Generate unique run ID
    if run_id is None:
        run_id = str(uuid.uuid4())
    
    if run_date is None:
        run_date = datetime.now().date()
    
    print(f"ETL Run ID: {run_id}, Date: {run_date}")
    
    # Check if this run already completed
    existing_run = engine.execute(f"""
        SELECT status FROM etl_runs 
        WHERE run_id = '{run_id}' AND run_date = '{run_date}'
    """).fetchone()
    
    if existing_run and existing_run[0] == 'completed':
        print(f"Run {run_id} already completed, skipping")
        return
    
    # Record run start
    engine.execute(f"""
        INSERT INTO etl_runs (run_id, run_date, status, started_at)
        VALUES ('{run_id}', '{run_date}', 'running', NOW())
        ON CONFLICT (run_id, run_date) DO UPDATE SET started_at = NOW()
    """)
    
    try:
        # Extract
        data = extract_from_api(run_date)
        
        # Transform (add run_id to each record)
        data['etl_run_id'] = run_id
        data['etl_loaded_at'] = datetime.now()
        
        # Load with merge (upsert)
        # Option 1: Delete + Insert (for small datasets)
        engine.execute(f"""
            DELETE FROM fact_orders WHERE order_date = '{run_date}'
        """)
        data.to_sql('fact_orders', engine, if_exists='append', index=False)
        
        # Record completion
        engine.execute(f"""
            UPDATE etl_runs 
            SET status = 'completed', completed_at = NOW(), rows_processed = {len(data)}
            WHERE run_id = '{run_id}' AND run_date = '{run_date}'
        """)
        
        print(f"ETL completed: {len(data)} rows processed")
        
    except Exception as e:
        # Record failure
        engine.execute(f"""
            UPDATE etl_runs 
            SET status = 'failed', error_message = '{str(e)}'
            WHERE run_id = '{run_id}' AND run_date = '{run_date}'
        """)
        raise

# Safe to retry with same run_id
daily_etl(run_date='2024-01-15', run_id='specific-run-id-123')

dbt Mart Model Example

-- models/marts/fact_orders.sql
{{
    config(
        materialized='table',
        cluster_by=['order_month', 'customer_id']
    )
}}

WITH orders AS (
    SELECT * FROM {{ ref('stg_orders') }}
),

customers AS (
    SELECT * FROM {{ ref('stg_customers') }}
),

order_items AS (
    SELECT 
        order_id,
        COUNT(*) AS item_count,
        SUM(quantity) AS total_quantity
    FROM {{ ref('stg_order_items') }}
    GROUP BY order_id
)

SELECT
    o.order_id,
    o.customer_id,
    c.customer_name,
    c.customer_segment,
    o.order_date,
    o.order_month,
    o.cleaned_total_amount AS total_amount,
    o.currency,
    o.standardized_status AS status,
    i.item_count,
    i.total_quantity,
    
    -- Business metrics
    CASE 
        WHEN o.cleaned_total_amount >= 1000 THEN 'High Value'
        WHEN o.cleaned_total_amount >= 100 THEN 'Medium Value'
        ELSE 'Low Value'
    END AS order_value_tier,
    
    o.dbt_loaded_at

FROM orders o
LEFT JOIN customers c ON o.customer_id = c.customer_id
LEFT JOIN order_items i ON o.order_id = i.order_id

SCD Type 2 Implementation

-- dbt snapshot for SCD2
{% snapshot users_snapshot %}

{{
    config(
      target_schema='snapshots',
      unique_key='user_id',
      strategy='check',
      check_cols=['address', 'email', 'status'],
    )
}}

SELECT * FROM {{ source('raw', 'users') }}

{% endsnapshot %}

Kafka Consumer with Error Handling

from kafka import KafkaConsumer, KafkaProducer
import json

consumer = KafkaConsumer(
    'orders',
    bootstrap_servers=['kafka:9092'],
    value_deserializer=lambda m: json.loads(m.decode('utf-8')),
    auto_offset_reset='earliest',
    enable_auto_commit=False,
    group_id='order-processor'
)

dlq_producer = KafkaProducer(
    bootstrap_servers=['kafka:9092'],
    value_serializer=lambda m: json.dumps(m).encode('utf-8')
)

def process_message(message):
    try:
        order = message.value
        
        # Validate required fields
        required_fields = ['order_id', 'customer_id', 'amount']
        for field in required_fields:
            if field not in order:
                raise ValueError(f"Missing required field: {field}")
        
        # Process the order
        result = transform_and_load(order)
        return result
        
    except Exception as e:
        # Send to Dead Letter Queue
        dlq_message = {
            'original_message': message.value,
            'error': str(e),
            'topic': message.topic,
            'partition': message.partition,
            'offset': message.offset,
            'timestamp': datetime.now().isoformat()
        }
        dlq_producer.send('orders-dlq', dlq_message)
        print(f"Sent to DLQ: {e}")
        return None

for message in consumer:
    result = process_message(message)
    if result:
        consumer.commit()

Spark Streaming with Watermarks

from pyspark.sql import SparkSession
from pyspark.sql.functions import window, col, from_json
from pyspark.sql.types import StructType, StringType, TimestampType, DoubleType

spark = SparkSession.builder \
    .appName("OrderStreaming") \
    .getOrCreate()

schema = StructType() \
    .add("order_id", StringType()) \
    .add("customer_id", StringType()) \
    .add("amount", DoubleType()) \
    .add("event_time", TimestampType())

# Read from Kafka
orders_stream = spark \
    .readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "kafka:9092") \
    .option("subscribe", "orders") \
    .load() \
    .select(from_json(col("value").cast("string"), schema).alias("data")) \
    .select("data.*")

# Aggregate with watermark for late data handling
order_aggregates = orders_stream \
    .withWatermark("event_time", "10 minutes") \
    .groupBy(
        window(col("event_time"), "5 minutes"),
        col("customer_id")
    ) \
    .agg(
        {"amount": "sum", "order_id": "count"}
    )

# Write to console (or sink to database)
query = order_aggregates \
    .writeStream \
    .outputMode("update") \
    .format("console") \
    .start()

query.awaitTermination()

Data Engineer - Technical Reference

Workflow: Build Batch ETL Pipeline with Airflow + dbt

Use case: Daily data sync from PostgreSQL → Snowflake for analytics

Step 1: Source System Analysis

# scripts/analyze_source.py
import pandas as pd
from sqlalchemy import create_engine

# Connect to source database
engine = create_engine('postgresql://user:pass@prod-db:5432/appdb')

# Analyze table sizes and row counts
tables_query = """
SELECT 
    schemaname,
    tablename,
    pg_size_pretty(pg_total_relation_size(schemaname||'.'||tablename)) AS size,
    (SELECT COUNT(*) FROM information_schema.columns WHERE table_name = tablename) AS columns
FROM pg_tables
WHERE schemaname NOT IN ('pg_catalog', 'information_schema')
ORDER BY pg_total_relation_size(schemaname||'.'||tablename) DESC
LIMIT 20;
"""

tables_df = pd.read_sql(tables_query, engine)
print(tables_df)

# Identify incremental load candidates (tables with updated_at column)
incremental_tables = pd.read_sql("""
SELECT table_name
FROM information_schema.columns
WHERE column_name IN ('updated_at', 'modified_at', 'last_modified')
GROUP BY table_name
HAVING COUNT(*) >= 1;
""", engine)

print(f"\nIncremental load candidates: {len(incremental_tables)} tables")

Step 2: Airflow DAG for Orchestration

# dags/daily_etl.py
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.providers.snowflake.operators.snowflake import SnowflakeOperator
from airflow.providers.postgres.hooks.postgres import PostgresHook
from airflow.providers.snowflake.hooks.snowflake import SnowflakeHook
from datetime import datetime, timedelta
import pandas as pd

default_args = {
    'owner': 'data-engineering',
    'depends_on_past': False,
    'email': ['data-alerts@company.com'],
    'email_on_failure': True,
    'email_on_retry': False,
    'retries': 3,
    'retry_delay': timedelta(minutes=5),
    'retry_exponential_backoff': True,
    'max_retry_delay': timedelta(minutes=30),
}

dag = DAG(
    'daily_etl_postgres_to_snowflake',
    default_args=default_args,
    description='Daily ETL from PostgreSQL to Snowflake',
    schedule_interval='0 2 * * *',  # 2 AM daily
    start_date=datetime(2024, 1, 1),
    catchup=False,
    tags=['etl', 'snowflake', 'production'],
)

def extract_and_load(table_name, incremental=False):
    """Extract from PostgreSQL and load to Snowflake"""
    
    pg_hook = PostgresHook(postgres_conn_id='postgres_prod')
    
    if incremental:
        sf_hook = SnowflakeHook(snowflake_conn_id='snowflake_warehouse')
        last_sync = sf_hook.get_first(f"""
            SELECT MAX(updated_at) 
            FROM ANALYTICS.RAW.{table_name.upper()}_SYNC_METADATA
        """)[0]
        
        query = f"""
            SELECT * FROM {table_name}
            WHERE updated_at > '{last_sync}'
        """
        print(f"Incremental load from {last_sync}")
    else:
        query = f"SELECT * FROM {table_name}"
        print(f"Full refresh of {table_name}")
    
    # Extract (stream in chunks for large tables)
    chunk_size = 100000
    df_iter = pg_hook.get_pandas_df_by_chunks(query, chunksize=chunk_size)
    
    sf_hook = SnowflakeHook(snowflake_conn_id='snowflake_warehouse')
    
    total_rows = 0
    for chunk_num, df_chunk in enumerate(df_iter):
        # Data quality checks
        assert df_chunk.isnull().sum().sum() < len(df_chunk) * 0.1, \
            f"Too many nulls in {table_name} chunk {chunk_num}"
        
        sf_hook.insert_rows(
            table=f'RAW.STG_{table_name.upper()}',
            rows=df_chunk.values.tolist(),
            target_fields=df_chunk.columns.tolist(),
        )
        
        total_rows += len(df_chunk)
        print(f"Loaded chunk {chunk_num}: {len(df_chunk)} rows")
    
    print(f"Total rows loaded: {total_rows}")
    return total_rows

# Create tasks dynamically for each table
tables_config = [
    {'name': 'users', 'incremental': False},
    {'name': 'orders', 'incremental': True},
    {'name': 'transactions', 'incremental': True},
    {'name': 'products', 'incremental': False},
]

extract_tasks = []
for table in tables_config:
    task = PythonOperator(
        task_id=f'extract_load_{table["name"]}',
        python_callable=extract_and_load,
        op_kwargs={
            'table_name': table['name'],
            'incremental': table['incremental']
        },
        dag=dag,
    )
    extract_tasks.append(task)

# dbt transformation task
dbt_run = SnowflakeOperator(
    task_id='dbt_transform',
    snowflake_conn_id='snowflake_warehouse',
    sql='CALL ANALYTICS.DBT.RUN_TRANSFORMATIONS()',
    dag=dag,
)

# Data quality checks
quality_checks = SnowflakeOperator(
    task_id='data_quality_checks',
    snowflake_conn_id='snowflake_warehouse',
    sql="""
        SELECT 'orders' AS table_name,
            COUNT(*) - COUNT(DISTINCT order_id) AS duplicates
        FROM ANALYTICS.MART.ORDERS
        HAVING duplicates > 0
        
        UNION ALL
        
        SELECT 'orders' AS table_name,
            COUNT(*) AS null_customer_ids
        FROM ANALYTICS.MART.ORDERS
        WHERE customer_id IS NULL
        HAVING null_customer_ids > 0
    """,
    dag=dag,
)

# DAG dependencies
extract_tasks >> dbt_run >> quality_checks

Step 3: dbt Transformations

-- models/staging/stg_orders.sql
{{
    config(
        materialized='incremental',
        unique_key='order_id',
        on_schema_change='append_new_columns',
        cluster_by=['order_date']
    )
}}

WITH source AS (
    SELECT *
    FROM {{ source('raw', 'orders') }}
    
    {% if is_incremental() %}
        WHERE updated_at > (SELECT MAX(updated_at) FROM {{ this }})
    {% endif %}
),

cleaned AS (
    SELECT
        order_id,
        customer_id,
        order_date,
        total_amount,
        currency,
        status,
        
        CASE 
            WHEN total_amount < 0 THEN 0
            ELSE total_amount
        END AS cleaned_total_amount,
        
        UPPER(TRIM(status)) AS standardized_status,
        DATE_TRUNC('month', order_date) AS order_month,
        updated_at,
        CURRENT_TIMESTAMP() AS dbt_loaded_at
    
    FROM source
    WHERE order_id IS NOT NULL
)

SELECT * FROM cleaned

Execution Results

# Run Airflow DAG
airflow dags trigger daily_etl_postgres_to_snowflake

# Expected output:
# [2024-01-15 02:00:00] extract_load_users: Loaded 10,000,000 rows (full refresh)
# [2024-01-15 02:15:00] extract_load_orders: Loaded 150,000 rows (incremental)
# [2024-01-15 02:20:00] extract_load_transactions: Loaded 500,000 rows (incremental)
# [2024-01-15 02:25:00] dbt_transform: 25 models run successfully
# [2024-01-15 02:30:00] data_quality_checks: PASSED (no issues detected)

# Cost analysis:
# - Snowflake compute: $45 (warehouse running 30 minutes)
# - Snowflake storage: $23/month (1TB compressed)
# - Total monthly cost: ~$2,000

Detailed Patterns

Pattern: Idempotent Partition Overwrite

# PySpark example: Overwrite partition based on execution date
def write_daily_partition(df, target_table, execution_date):
    """
    Writes data to a specific partition, overwriting existing data for that day.
    Safe to re-run multiple times.
    """
    (df
     .write
     .mode("overwrite")
     .partitionBy("process_date")
     .option("partitionOverwriteMode", "dynamic")
     .format("parquet")
     .saveAsTable(target_table))

# SQL Equivalent (Snowflake/BigQuery)
"""
MERGE INTO target t
USING source s
ON t.id = s.id AND t.process_date = s.process_date
WHEN MATCHED THEN
  UPDATE SET ...
WHEN NOT MATCHED THEN
  INSERT ...
"""

Pattern: Data Quality Circuit Breaker

def check_data_quality(df, thresholds):
    """
    Returns True if data quality is acceptable, False otherwise.
    """
    total_rows = df.count()
    null_count = df.filter(col("critical_column").isNull()).count()
    null_ratio = null_count / total_rows
    
    if null_ratio > thresholds['max_null_ratio']:
        raise DataQualityException(
            f"Null ratio {null_ratio:.2%} exceeds limit {thresholds['max_null_ratio']:.2%}"
        )
        
    unique_count = df.select("id").distinct().count()
    if unique_count != total_rows:
        raise DataQualityException(
            f"Duplicate IDs found: {total_rows - unique_count} duplicates"
        )
        
    return True

# Usage in Airflow
check_task = PythonOperator(
    task_id='dq_check',
    python_callable=check_data_quality,
    op_kwargs={'thresholds': {'max_null_ratio': 0.01}}
)

Pattern: Dead Letter Queue (DLQ) for Streaming

# Flink Side Output for DLQ
process_stream = stream.process(ProcessFunction())
valid_data = process_stream.get_side_output("valid")
bad_data = process_stream.get_side_output("dlq")

# Sink bad data to S3/GCS for manual inspection
bad_data.add_sink(
    FileSink.for_row_format(
        BasePath("s3://data-lake/dlq/"),
        Encoder()
    ).build()
)