Data Engineer

When NOT to Use

• For statistical analysis, dashboards, or insight generation — use data-analyst instead
• For ML model training, feature engineering, or model serving — use ml-engineer
• For database schema migrations in an application database (OLTP) — use database-migrations
• For business intelligence tool configuration (Tableau, Looker, Metabase) — use data-analyst
• Do not use this skill when the output is a chart or a recommendation; pipelines are the output here

Core Principles

1. Idempotency is non-negotiable. Every pipeline run must produce the same result whether it runs once or ten times. Design for re-runs, not happy paths. If a job fails halfway, re-running it must not create duplicates or corrupt state.
2. Quality gates before load. Data quality checks run in the Transform stage, before data lands in the destination. Never load dirty data and plan to clean it later — it never gets cleaned later.
3. Schema evolution is inevitable. Design schemas and models to tolerate additive changes (new columns) gracefully. Non-additive changes (type changes, column removals) require explicit migration protocols with backward-compatible transitional periods.
4. Fail loudly on data anomalies. A pipeline that silently passes bad data is more dangerous than one that fails. Use hard assertions for critical constraints (non-null primary keys, referential integrity). Use warnings for soft constraints (unexpected null rates, cardinality drops).
5. Freshness is a first-class SLO. Define maximum acceptable data latency for every dataset. Monitor it. Alert when data is stale. "The pipeline ran successfully" is not the same as "the data is fresh."
6. Prefer incremental over full refresh. Full table refreshes are expensive and operationally fragile at scale. Default to incremental loading with proper watermarking or CDC patterns.
7. Lineage is documentation. Instrument pipelines to emit lineage metadata (source → transform → destination). Undocumented lineage means analysts cannot trust the data and cannot debug when it breaks.

Pipeline Architecture Patterns

Batch ELT Pattern (dbt + Warehouse)

Source Systems
    │
    ▼ (raw extract — append-only, never modify)
Raw Layer (warehouse: raw.source_name.table)
    │
    ▼ (dbt staging models — rename, cast, deduplicate)
Staging Layer (warehouse: staging.stg_source__table)
    │
    ▼ (dbt intermediate models — join, enrich)
Intermediate Layer (warehouse: intermediate.int_*)
    │
    ▼ (dbt mart models — business-facing aggregations)
Mart Layer (warehouse: marts.dim_* / fact_*)
    │
    ▼
Analysts / BI Tools

Key rule: Raw layer is append-only. Never modify raw data. If a source record changes, append the new version with a loaded_at timestamp.

Streaming Pipeline Pattern (Kafka + Flink)

Source (DB CDC / API / Events)
    │
    ▼ Kafka Topic (raw events, 7-day retention)
    │
    ▼ Flink Job (stateful processing)
    │   ├── Deduplication (by event_id, tumbling window)
    │   ├── Schema validation (Avro/Protobuf registry check)
    │   ├── Enrichment (lookup join against dimension tables)
    │   └── Aggregation (windowed metrics)
    │
    ▼ Kafka Topic (processed events)
    │
    ▼ Sink (warehouse, search index, cache)

Incremental Loading Patterns

Watermark-Based Incremental (dbt)

-- models/staging/stg_orders.sql
{{
  config(
    materialized='incremental',
    unique_key='order_id',
    on_schema_change='sync_all_columns',
    incremental_strategy='merge'
  )
}}

SELECT
  order_id,
  user_id,
  status,
  total_amount,
  created_at,
  updated_at,
  CURRENT_TIMESTAMP AS _dbt_loaded_at
FROM {{ source('raw', 'orders') }}

{% if is_incremental() %}
  -- Only process records updated since last run
  -- Use a 1-hour lookback to handle late-arriving data
  WHERE updated_at >= (SELECT MAX(updated_at) - INTERVAL '1 hour' FROM {{ this }})
{% endif %}

Change Data Capture (CDC) Pattern

# cdc_processor.py — Debezium → Kafka → Warehouse
from dataclasses import dataclass
from enum import Enum

class CDCOperation(Enum):
    INSERT = "c"   # Debezium: create
    UPDATE = "u"   # Debezium: update
    DELETE = "d"   # Debezium: delete

@dataclass
class CDCRecord:
    operation: CDCOperation
    source_table: str
    primary_key: dict
    before: dict | None   # None for inserts
    after: dict | None    # None for deletes
    transaction_timestamp: str

def apply_cdc_record(record: CDCRecord, target_table) -> None:
    """
    Apply a CDC record idempotently.
    UPSERT for inserts/updates, soft-delete for deletes.
    Never hard-delete — use is_deleted flag + deleted_at timestamp.
    """
    if record.operation == CDCOperation.DELETE:
        target_table.upsert({
            **record.primary_key,
            "is_deleted": True,
            "deleted_at": record.transaction_timestamp
        })
    else:
        target_table.upsert({
            **record.after,
            "is_deleted": False,
            "_source_updated_at": record.transaction_timestamp
        })

Data Quality Framework

Great Expectations Integration Pattern

import great_expectations as ge
from great_expectations.core import ExpectationSuite

def build_orders_expectation_suite() -> ExpectationSuite:
    """
    Define data quality contract for the orders table.
    Hard assertions: pipeline fails if violated.
    Soft assertions: logged as warnings, pipeline continues.
    """
    context = ge.get_context()
    suite = context.create_expectation_suite("orders.critical")

    # HARD: Primary key integrity
    suite.add_expectation(
        ge.expectations.ExpectColumnValuesToBeUnique(column="order_id")
    )
    suite.add_expectation(
        ge.expectations.ExpectColumnValuesToNotBeNull(column="order_id")
    )

    # HARD: Referential integrity
    suite.add_expectation(
        ge.expectations.ExpectColumnValuesToNotBeNull(column="user_id")
    )

    # HARD: Value constraints
    suite.add_expectation(
        ge.expectations.ExpectColumnValuesToBeBetween(
            column="total_amount", min_value=0, max_value=100_000
        )
    )
    suite.add_expectation(
        ge.expectations.ExpectColumnValuesToBeInSet(
            column="status",
            value_set=["pending", "processing", "shipped", "delivered", "cancelled", "refunded"]
        )
    )

    # SOFT: Freshness check (warn if no records in last 2 hours)
    suite.add_expectation(
        ge.expectations.ExpectTableRowCountToBeGreaterThan(value=0)
    )

    return suite

def run_quality_gate(df, suite_name: str, fail_on_critical: bool = True) -> dict:
    """
    Run quality checks. Fails pipeline on critical violations.
    Returns quality report for logging/alerting.
    """
    context = ge.get_context()
    validator = context.get_validator(batch_request=..., expectation_suite_name=suite_name)
    results = validator.validate()

    failed = [r for r in results.results if not r.success]
    critical_failures = [r for r in failed if r.expectation_config.kwargs.get("severity") != "warn"]

    if fail_on_critical and critical_failures:
        raise DataQualityError(
            f"Pipeline halted: {len(critical_failures)} critical quality violations.\n" +
            "\n".join(str(r.expectation_config) for r in critical_failures)
        )

    return {
        "total_checks": len(results.results),
        "passed": results.statistics["successful_expectations"],
        "failed": results.statistics["unsuccessful_expectations"],
        "critical_failures": len(critical_failures)
    }

Schema Evolution Strategies

Safe Migration Protocol

┌──────────────────────────────────────────────────────────────┐
│                  Schema Change Classification                 │
│                                                              │
│  ADDITIVE (safe, non-breaking):                              │
│    + Add nullable column    → Apply immediately              │
│    + Add new table          → Apply immediately              │
│    + Widen VARCHAR length   → Apply immediately              │
│                                                              │
│  NON-ADDITIVE (breaking, requires protocol):                 │
│    - Remove column          → Deprecate first (30 days)      │
│    - Rename column          → Add alias, migrate, remove old │
│    - Change column type     → Add new column, backfill, swap │
│    - Change primary key     → Major migration protocol       │
└──────────────────────────────────────────────────────────────┘

Column Rename Protocol (dbt)

-- Step 1: Add new column alongside old (deploy, run pipeline)
ALTER TABLE orders ADD COLUMN customer_id BIGINT;
UPDATE orders SET customer_id = user_id;  -- backfill

-- Step 2: Update all dbt models to use new column
-- models/staging/stg_orders.sql
SELECT
  order_id,
  COALESCE(customer_id, user_id) AS customer_id,  -- transitional alias
  -- ...
FROM raw.orders

-- Step 3: After 30 days, confirm no consumers reference user_id
-- Step 4: Drop old column
ALTER TABLE orders DROP COLUMN user_id;

Pipeline Observability

Freshness Monitoring

from datetime import datetime, timedelta
from dataclasses import dataclass

@dataclass
class FreshnessContract:
    table: str
    max_staleness_hours: float
    alert_channel: str

FRESHNESS_CONTRACTS = [
    FreshnessContract("fact_orders", max_staleness_hours=1.0, alert_channel="#data-alerts"),
    FreshnessContract("fact_events", max_staleness_hours=0.25, alert_channel="#data-alerts"),
    FreshnessContract("dim_users", max_staleness_hours=24.0, alert_channel="#data-alerts"),
]

def check_freshness(contract: FreshnessContract, warehouse) -> dict:
    """Check if a table has been updated within its SLO."""
    result = warehouse.query(f"""
        SELECT MAX(_dbt_loaded_at) AS last_updated
        FROM {contract.table}
    """).fetchone()

    last_updated = result["last_updated"]
    staleness_hours = (datetime.utcnow() - last_updated).total_seconds() / 3600

    return {
        "table": contract.table,
        "last_updated": last_updated.isoformat(),
        "staleness_hours": round(staleness_hours, 2),
        "slo_hours": contract.max_staleness_hours,
        "status": "FRESH" if staleness_hours <= contract.max_staleness_hours else "STALE",
        "alert_required": staleness_hours > contract.max_staleness_hours
    }

Data Lineage Metadata Emission

def emit_lineage_event(
    source_tables: list[str],
    destination_table: str,
    job_name: str,
    row_count: int,
    run_id: str
) -> None:
    """
    Emit lineage metadata for every pipeline run.
    Consumed by data catalog (DataHub, OpenMetadata, Marquez).
    """
    lineage_event = {
        "eventType": "COMPLETE",
        "run": {"runId": run_id},
        "job": {"namespace": "data_platform", "name": job_name},
        "inputs": [{"namespace": "warehouse", "name": t} for t in source_tables],
        "outputs": [{"namespace": "warehouse", "name": destination_table}],
        "outputFacets": {
            "rowCount": {"_producer": job_name, "rowCount": row_count}
        }
    }
    lineage_client.emit(lineage_event)

Idempotency Checklist

Before deploying any pipeline, verify:

• Re-running the job twice produces identical destination state (not double the rows)
• If the job fails mid-run and is re-run, the output is consistent with a clean run
• Incremental logic uses MERGE/UPSERT, not INSERT only
• Watermark query includes a lookback buffer (1 hour minimum) for late-arriving data
• CDC deletes are soft-deletes (is_deleted flag), not hard-deletes
• Job run IDs are logged and can be used to trace which run produced which data

Self-Verification Checklist

Before declaring a pipeline complete:

• Data quality suite is defined and all critical expectations pass on the current data
• Idempotency test: ran the pipeline twice — row counts are identical, no duplicate primary keys
• Schema evolution test: added a nullable column to the source — pipeline handled it without failure
• Freshness contract is defined and monitored (staleness alert fires if data is >SLO threshold)
• Lineage metadata is emitted on every run (source → destination relationships are visible in catalog)
• Incremental logic is verified: re-run with is_incremental=True processes only new/changed records
• Pipeline is observable: run duration, row count, and quality score are logged and queryable

Success Criteria

Task is complete when:

1. Pipeline runs to completion with exit code 0 and row count > 0 in the destination
2. All critical data quality expectations pass (0 critical failures)
3. Idempotency verified: two consecutive runs produce identical row counts and checksums
4. Freshness contract deployed: check_freshness() returns FRESH status within 10 minutes of job completion
5. No duplicate primary keys exist in the destination table after a full pipeline run + re-run
6. Schema evolution test passes: adding a new nullable column to the source does not fail the pipeline

Anti-Patterns

• Never use INSERT without a deduplication strategy because duplicate records accumulate silently across pipeline reruns, and downstream analysts will eventually aggregate over inflated counts without knowing they are double-counting, producing reports that overstate revenue, events, or user activity by an arbitrary factor — always MERGE/UPSERT or deduplicate in staging.
• Never hard-delete records from a data warehouse because permanent deletion destroys the audit trail required for compliance, makes point-in-time analysis impossible, and breaks any downstream model that references the deleted key — use a soft-delete with is_deleted flag and deleted_at timestamp instead.
• Never load data to a destination before running quality checks because loading dirty data and planning to clean it later means it never gets cleaned, analysts build dashboards on corrupt records, and fixing the data retroactively requires a full historical reload that can take hours or days.
• Never use a full table refresh where incremental loading is viable because full refreshes fail at scale when the source table exceeds memory or query timeout limits, and a failed mid-refresh leaves the destination in a partially overwritten state that analysts are still querying.
• Never deploy a pipeline without a freshness SLO because without a defined maximum acceptable data latency, stakeholders have no signal when data is stale — a pipeline that ran at 2 AM but silently stopped producing records by 6 AM will go undetected until an analyst acts on hours-old data.
• Never alter a non-nullable column type directly because an in-place type change locks the table during migration, breaks all downstream consumers the moment the pipeline restarts, and cannot be rolled back without restoring from backup — use the additive migration protocol: add new column, backfill, swap references, then drop old column after a deprecation window.
• Never silence a data quality failure with a try/except without alerting because swallowing a quality exception allows a corrupt batch to land in the destination silently, leaving analysts to discover data anomalies hours later when the business impact has already propagated to reports and dashboards.

Failure Modes

Integration with Mega-Mind

The data engineer skill is the infrastructure layer in the Data vertical:

data-engineer (build reliable pipelines) → data-analyst (analyze trustworthy data) → ml-engineer (model and predict)

• Pair with observability-specialist to wire pipeline metrics into monitoring dashboards and SLO alerts
• Use database-migrations for schema changes in application databases (OLTP); this skill handles OLAP/warehouse schema changes
• Coordinate with infra-architect for warehouse provisioning, IAM policies, and network access patterns
• When a pipeline is mature and analysts need to explore it, hand off to data-analyst for EDA and reporting