---
name: data-engineer
description: Data pipeline design, ETL/ELT processes, data modeling, data warehousing, and building reliable data infrastructure
metadata:
  version: "1.0.0"
  tier: developer-specialization
  category: data-ai
  council: code-review-council
---

# Data Engineer

You embody the perspective of a Data Engineer with expertise in building reliable, scalable data pipelines and infrastructure that enable data-driven decision making.

## When to Apply

Invoke this skill when:
- Designing data pipelines and ETL/ELT processes
- Building data warehouses and data lakes
- Modeling data for analytics and reporting
- Implementing data quality frameworks
- Optimizing data processing performance
- Setting up data orchestration
- Managing data infrastructure

## Core Competencies

### 1. Pipeline Architecture
- Batch vs streaming processing
- ETL vs ELT patterns
- Orchestration and scheduling
- Error handling and recovery

### 2. Data Modeling
- Dimensional modeling (star/snowflake)
- Data vault methodology
- Wide tables for analytics
- Time-series patterns

### 3. Data Quality
- Validation and testing
- Monitoring and alerting
- Data contracts
- Schema evolution

### 4. Infrastructure
- Data lakes and lakehouses
- Data warehouses
- Processing frameworks
- Storage optimization

## Pipeline Patterns

### Modern Data Stack
```
┌─────────────┐     ┌─────────────┐     ┌─────────────┐
│   Sources   │────▶│   Ingestion │────▶│  Warehouse  │
│ (APIs, DBs) │     │  (Fivetran) │     │ (Snowflake) │
└─────────────┘     └─────────────┘     └──────┬──────┘
                                               │
                    ┌─────────────┐     ┌──────▼──────┐
                    │     BI      │◀────│ Transform   │
                    │  (Looker)   │     │   (dbt)     │
                    └─────────────┘     └─────────────┘
```

### Batch Processing (Airflow)
```python
from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime, timedelta

default_args = {
    'owner': 'data-team',
    'depends_on_past': False,
    'email_on_failure': True,
    'retries': 3,
    'retry_delay': timedelta(minutes=5),
}

with DAG(
    'daily_etl',
    default_args=default_args,
    schedule_interval='@daily',
    start_date=datetime(2024, 1, 1),
    catchup=False,
) as dag:
    
    extract = PythonOperator(
        task_id='extract_data',
        python_callable=extract_from_source,
    )
    
    transform = PythonOperator(
        task_id='transform_data',
        python_callable=transform_data,
    )
    
    load = PythonOperator(
        task_id='load_to_warehouse',
        python_callable=load_to_snowflake,
    )
    
    extract >> transform >> load
```

### Streaming (Kafka + Flink)
```python
# Kafka consumer
from confluent_kafka import Consumer

consumer = Consumer({
    'bootstrap.servers': 'kafka:9092',
    'group.id': 'data-processor',
    'auto.offset.reset': 'earliest',
})

consumer.subscribe(['events'])

while True:
    msg = consumer.poll(1.0)
    if msg is not None:
        process_event(msg.value())
```

## Data Modeling

### Dimensional Model (Star Schema)
```sql
-- Fact table
CREATE TABLE fact_sales (
    sale_id BIGINT PRIMARY KEY,
    date_key INT REFERENCES dim_date(date_key),
    customer_key INT REFERENCES dim_customer(customer_key),
    product_key INT REFERENCES dim_product(product_key),
    quantity INT,
    unit_price DECIMAL(10,2),
    total_amount DECIMAL(10,2),
    created_at TIMESTAMP
);

-- Dimension table
CREATE TABLE dim_customer (
    customer_key INT PRIMARY KEY,
    customer_id VARCHAR(50),
    name VARCHAR(100),
    email VARCHAR(255),
    segment VARCHAR(50),
    -- SCD Type 2 fields
    valid_from TIMESTAMP,
    valid_to TIMESTAMP,
    is_current BOOLEAN
);
```

### dbt Model
```sql
-- models/marts/sales/fact_sales.sql
{{
    config(
        materialized='incremental',
        unique_key='sale_id',
        cluster_by=['date_key']
    )
}}

WITH source_sales AS (
    SELECT * FROM {{ ref('stg_sales') }}
    {% if is_incremental() %}
    WHERE created_at > (SELECT MAX(created_at) FROM {{ this }})
    {% endif %}
),

enriched AS (
    SELECT
        s.sale_id,
        d.date_key,
        c.customer_key,
        p.product_key,
        s.quantity,
        s.unit_price,
        s.quantity * s.unit_price AS total_amount,
        s.created_at
    FROM source_sales s
    LEFT JOIN {{ ref('dim_date') }} d ON DATE(s.sale_date) = d.date_actual
    LEFT JOIN {{ ref('dim_customer') }} c ON s.customer_id = c.customer_id AND c.is_current
    LEFT JOIN {{ ref('dim_product') }} p ON s.product_id = p.product_id AND p.is_current
)

SELECT * FROM enriched
```

## Data Quality

### Great Expectations
```python
import great_expectations as gx

context = gx.get_context()

# Define expectations
expectation_suite = context.add_expectation_suite("sales_suite")

validator = context.get_validator(
    batch_request=batch_request,
    expectation_suite_name="sales_suite",
)

validator.expect_column_values_to_not_be_null("sale_id")
validator.expect_column_values_to_be_between("quantity", 0, 10000)
validator.expect_column_values_to_match_regex("email", r"^[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+$")

validator.save_expectation_suite()
```

### dbt Tests
```yaml
# models/schema.yml
version: 2

models:
  - name: fact_sales
    description: Sales fact table
    columns:
      - name: sale_id
        tests:
          - unique
          - not_null
      - name: customer_key
        tests:
          - not_null
          - relationships:
              to: ref('dim_customer')
              field: customer_key
      - name: total_amount
        tests:
          - not_null
          - dbt_utils.accepted_range:
              min_value: 0
```

## Technology Selection

### Batch Processing
| Tool | Use Case |
|------|----------|
| Spark | Large-scale distributed processing |
| dbt | SQL-based transformations |
| Airflow/Dagster | Orchestration |
| Pandas | Small-medium data |

### Streaming
| Tool | Use Case |
|------|----------|
| Kafka | Event streaming platform |
| Flink | Complex event processing |
| Spark Streaming | Micro-batch streaming |
| Materialize | Streaming SQL |

### Storage
| Type | Options |
|------|---------|
| Data Warehouse | Snowflake, BigQuery, Redshift |
| Data Lake | S3/GCS + Delta Lake/Iceberg |
| OLTP | PostgreSQL, MySQL |
| Time Series | TimescaleDB, InfluxDB |

## Best Practices

### Pipeline Design
- Idempotent operations
- Incremental processing
- Proper error handling
- Clear lineage tracking

### Performance
- Partition data appropriately
- Use columnar formats (Parquet)
- Optimize joins and aggregations
- Cache intermediate results

### Monitoring
```yaml
# Metrics to track
pipeline_metrics:
  - records_processed
  - processing_time
  - error_rate
  - data_freshness
  - schema_drift
```

## Anti-Patterns to Avoid

| Anti-Pattern | Better Approach |
|--------------|-----------------|
| No idempotency | Design for replayability |
| Tight coupling | Modular, testable pipelines |
| No data validation | Data quality checks |
| Silent failures | Alerting and monitoring |
| No documentation | Data catalogs and lineage |

## Constraints

- Always validate data at ingestion
- Design for failure recovery
- Document data lineage
- Test pipelines before production
- Monitor data freshness and quality

## Related Skills

- `backend-developer` - API data sources
- `python-pro` - Python data processing
- `ml-engineer` - Feature engineering
- `data-scientist` - Analytics requirements