v1.0.0 to v1.1.0

- Data engineering skill for building scalable data pipelines, ETL/ELT systems,

- and data infrastructure. Expertise in Python, SQL, Spark, Airflow, dbt, Kafka,

- and modern data stack. Includes data modeling, pipeline orchestration, data

- quality, and DataOps. Use when designing data architectures, building data

- pipelines, optimizing data workflows, implementing data governance, or

- troubleshooting data issues.

- version: 1.0.0

- updated: 2026-03-31

- Production-grade data engineering skill for building scalable, reliable data systems.

-

- ## Table of Contents

-

- 1. [Trigger Phrases](#trigger-phrases)

- 2. [Quick Start](#quick-start)

- 3. [Workflows](#workflows)

- - [Building a Batch ETL Pipeline](#workflow-1-building-a-batch-etl-pipeline)

- - [Implementing Real-Time Streaming](#workflow-2-implementing-real-time-streaming)

- - [Data Quality Framework Setup](#workflow-3-data-quality-framework-setup)

- 4. [Architecture Decision Framework](#architecture-decision-framework)

- 5. [Tech Stack](#tech-stack)

- 6. [Reference Documentation](#reference-documentation)

- 7. [Troubleshooting](#troubleshooting)

-

- ---

-

- ## Trigger Phrases

-

- Activate this skill when you see:

-

- **Pipeline Design:**

- - "Design a data pipeline for..."

- - "Build an ETL/ELT process..."

- - "How should I ingest data from..."

- - "Set up data extraction from..."

-

- **Architecture:**

- - "Should I use batch or streaming?"

- - "Lambda vs Kappa architecture"

- - "How to handle late-arriving data"

- - "Design a data lakehouse"

-

- **Data Modeling:**

- - "Create a dimensional model..."

- - "Star schema vs snowflake"

- - "Implement slowly changing dimensions"

- - "Design a data vault"

-

- **Data Quality:**

- - "Add data validation to..."

- - "Set up data quality checks"

- - "Monitor data freshness"

- - "Implement data contracts"

-

- **Performance:**

- - "Optimize this Spark job"

- - "Query is running slow"

- - "Reduce pipeline execution time"

- - "Tune Airflow DAG"

- ### Core Tools

-

- # Generate pipeline orchestration config

- --type airflow \

- --source postgres \

- --destination snowflake \

- --schedule "0 5 * * *"

- # Validate data quality

- python scripts/data_quality_validator.py validate \

- --input data/sales.parquet \

- --schema schemas/sales.json \

- --checks freshness,completeness,uniqueness

- # Optimize ETL performance

- python scripts/etl_performance_optimizer.py analyze \

- --query queries/daily_aggregation.sql \

- --engine spark \

- --recommend

- ```

- ---

- ## Workflows

- ### Workflow 1: Building a Batch ETL Pipeline

- **Scenario:** Extract data from PostgreSQL, transform with dbt, load to Snowflake.

- #### Step 1: Define Source Schema

- ```sql

- -- Document source tables

- SELECT

- table_name,

- column_name,

- data_type,

- is_nullable

- FROM information_schema.columns

- WHERE table_schema = 'source_schema'

- ORDER BY table_name, ordinal_position;

- ```

- #### Step 2: Generate Extraction Config

- --type airflow \

- --source postgres \

- --tables orders,customers,products \

- --mode incremental \

- --watermark updated_at \

- --output dags/extract_source.py

- #### Step 3: Create dbt Models

- ),

-

- renamed AS (

- SELECT

- order_id,

- customer_id,

- order_date,

- total_amount,

- status,

- _extracted_at

- FROM source

- WHERE order_date >= DATEADD(day, -3, CURRENT_DATE)

-

- SELECT * FROM renamed

- {{

- config(

- materialized='incremental',

- unique_key='order_id',

- cluster_by=['order_date']

- )

- }}

- SELECT

- o.order_id,

- o.customer_id,

- c.customer_segment,

- o.order_date,

- o.total_amount,

- o.status

- LEFT JOIN {{ ref('dim_customers') }} c

- ON o.customer_id = c.customer_id

-

- #### Step 4: Configure Data Quality Tests

-

- ```yaml

- # models/marts/schema.yml

- version: 2

-

- models:

- - name: fct_orders

- description: "Order fact table"

- columns:

- - name: order_id

- tests:

- - unique

- - not_null

- - name: total_amount

- tests:

- - not_null

- - dbt_utils.accepted_range:

- min_value: 0

- max_value: 1000000

- - name: order_date

- tests:

- - not_null

- - dbt_utils.recency:

- datepart: day

- field: order_date

- interval: 1

- ```

-

- #### Step 5: Create Airflow DAG

- # dags/daily_etl.py

- from airflow import DAG

- from airflow.providers.postgres.operators.postgres import PostgresOperator

- from airflow.operators.bash import BashOperator

- from airflow.utils.dates import days_ago

- from datetime import timedelta

-

- default_args = {

- 'owner': 'data-team',

- 'depends_on_past': False,

- 'email_on_failure': True,

- 'email': ['data-alerts@company.com'],

- 'retries': 2,

- 'retry_delay': timedelta(minutes=5),

- }

-

- with DAG(

- 'daily_etl_pipeline',

- default_args=default_args,

- description='Daily ETL from PostgreSQL to Snowflake',

- schedule_interval='0 5 * * *',

- start_date=days_ago(1),

- catchup=False,

- tags=['etl', 'daily'],

- ) as dag:

-

- extract = BashOperator(

- task_id='extract_source_data',

- bash_command='python /opt/airflow/scripts/extract.py --date {{ ds }}',

- )

-

- transform = BashOperator(

- task_id='run_dbt_models',

- bash_command='cd /opt/airflow/dbt && dbt run --select marts.*',

- )

-

- test = BashOperator(

- task_id='run_dbt_tests',

- bash_command='cd /opt/airflow/dbt && dbt test --select marts.*',

- )

-

- notify = BashOperator(

- task_id='send_notification',

- bash_command='python /opt/airflow/scripts/notify.py --status success',

- trigger_rule='all_success',

- )

-

- extract >> transform >> test >> notify

- #### Step 6: Validate Pipeline

- # Test locally

- dbt run --select stg_orders fct_orders

- dbt test --select fct_orders

-

- # Validate data quality

- python scripts/data_quality_validator.py validate \

- --table fct_orders \

- --checks all \

- --output reports/quality_report.json

- ---

-

- ### Workflow 2: Implementing Real-Time Streaming

-

- **Scenario:** Stream events from Kafka, process with Flink/Spark Streaming, sink to data lake.

- #### Step 1: Define Event Schema

- ```json

- {

- "$schema": "http://json-schema.org/draft-07/schema#",

- "title": "UserEvent",

- "type": "object",

- "required": ["event_id", "user_id", "event_type", "timestamp"],

- "properties": {

- "event_id": {"type": "string", "format": "uuid"},

- "user_id": {"type": "string"},

- "event_type": {"type": "string", "enum": ["page_view", "click", "purchase"]},

- "timestamp": {"type": "string", "format": "date-time"},

- "properties": {"type": "object"}

- }

- }

- ```

- #### Step 2: Create Kafka Topic

- # Create topic with appropriate partitions

- kafka-topics.sh --create \

- --bootstrap-server localhost:9092 \

- --topic user-events \

- --partitions 12 \

- --replication-factor 3 \

- --config retention.ms=604800000 \

- --config cleanup.policy=delete

-

- # Verify topic

- kafka-topics.sh --describe \

- --bootstrap-server localhost:9092 \

- --topic user-events

- #### Step 3: Implement Spark Streaming Job

- # streaming/user_events_processor.py

- from pyspark.sql import SparkSession

- from pyspark.sql.functions import (

- from_json, col, window, count, avg,

- to_timestamp, current_timestamp

- )

- from pyspark.sql.types import (

- StructType, StructField, StringType,

- TimestampType, MapType

- )

-

- # Initialize Spark

- spark = SparkSession.builder \

- .appName("UserEventsProcessor") \

- .config("spark.sql.streaming.checkpointLocation", "/checkpoints/user-events") \

- .config("spark.sql.shuffle.partitions", "12") \

- .getOrCreate()

-

- # Define schema

- event_schema = StructType([

- StructField("event_id", StringType(), False),

- StructField("user_id", StringType(), False),

- StructField("event_type", StringType(), False),

- StructField("timestamp", StringType(), False),

- StructField("properties", MapType(StringType(), StringType()), True)

- ])

-

- # Read from Kafka

- events_df = spark.readStream \

- .format("kafka") \

- .option("startingOffsets", "latest") \

- .option("failOnDataLoss", "false") \

- .load()

- # Parse JSON

- parsed_df = events_df \

- .select(from_json(col("value").cast("string"), event_schema).alias("data")) \

- .select("data.*") \

- .withColumn("event_timestamp", to_timestamp(col("timestamp")))

- # Windowed aggregation

- .groupBy(

- window(col("event_timestamp"), "5 minutes"),

- col("event_type")

- ) \

- .agg(

- count("*").alias("event_count"),

- approx_count_distinct("user_id").alias("unique_users")

- )

-

- # Write to Delta Lake

- query = aggregated_df.writeStream \

- .format("delta") \

- .outputMode("append") \

- .option("checkpointLocation", "/checkpoints/user-events-aggregated") \

- .option("path", "/data/lake/user_events_aggregated") \

- .trigger(processingTime="1 minute") \

- .start()

-

- query.awaitTermination()

- ```

-

- #### Step 4: Handle Late Data and Errors

-

- ```python

- # Dead letter queue for failed records

- from pyspark.sql.functions import current_timestamp, lit

-

- def process_with_error_handling(batch_df, batch_id):

- try:

- # Attempt processing

- valid_df = batch_df.filter(col("event_id").isNotNull())

- invalid_df = batch_df.filter(col("event_id").isNull())

-

- # Write valid records

- valid_df.write \

- .format("delta") \

- .mode("append") \

- .save("/data/lake/user_events")

-

- # Write invalid to DLQ

- if invalid_df.count() > 0:

- invalid_df \

- .withColumn("error_timestamp", current_timestamp()) \

- .withColumn("error_reason", lit("missing_event_id")) \

- .write \

- .format("delta") \

- .mode("append") \

- .save("/data/lake/dlq/user_events")

-

- except Exception as e:

- # Log error, alert, continue

- logger.error(f"Batch {batch_id} failed: {e}")

- raise

- # Use foreachBatch for custom processing

- query = parsed_df.writeStream \

- .foreachBatch(process_with_error_handling) \

- .start()

- #### Step 5: Monitor Stream Health

- # monitoring/stream_metrics.py

- from prometheus_client import Gauge, Counter, start_http_server

-

- # Define metrics

- RECORDS_PROCESSED = Counter(

- 'stream_records_processed_total',

- 'Total records processed',

- ['stream_name', 'status']

- )

-

- PROCESSING_LAG = Gauge(

- 'stream_processing_lag_seconds',

- 'Current processing lag',

- ['stream_name']

- )

-

- BATCH_DURATION = Gauge(

- 'stream_batch_duration_seconds',

- 'Last batch processing duration',

- ['stream_name']

- )

-

- def emit_metrics(query):

- """Emit Prometheus metrics from streaming query."""

- progress = query.lastProgress

- if progress:

- RECORDS_PROCESSED.labels(

- stream_name='user-events',

- status='success'

- ).inc(progress['numInputRows'])

-

- if progress['sources']:

- # Calculate lag from latest offset

- for source in progress['sources']:

- end_offset = source.get('endOffset', {})

- # Parse Kafka offsets and calculate lag

- ---

- ### Workflow 3: Data Quality Framework Setup

- **Scenario:** Implement comprehensive data quality monitoring with Great Expectations.

- #### Step 1: Initialize Great Expectations

- # Install and initialize

- pip install great_expectations

-

- great_expectations init

-

- # Connect to data source

- great_expectations datasource new

- #### Step 2: Create Expectation Suite

-

- ```python

- # expectations/orders_suite.py

- import great_expectations as gx

-

- context = gx.get_context()

-

- # Create expectation suite

- suite = context.add_expectation_suite("orders_quality_suite")

-

- # Add expectations

- validator = context.get_validator(

- batch_request={

- "datasource_name": "warehouse",

- "data_asset_name": "orders",

- },

- expectation_suite_name="orders_quality_suite"

- )

-

- # Schema expectations

- validator.expect_table_columns_to_match_ordered_list(

- column_list=[

- "order_id", "customer_id", "order_date",

- "total_amount", "status", "created_at"

- ]

- )

-

- # Completeness expectations

- validator.expect_column_values_to_not_be_null("order_id")

- validator.expect_column_values_to_not_be_null("customer_id")

- validator.expect_column_values_to_not_be_null("order_date")

-

- # Uniqueness expectations

- validator.expect_column_values_to_be_unique("order_id")

-

- # Range expectations

- validator.expect_column_values_to_be_between(

- "total_amount",

- min_value=0,

- max_value=1000000

- )

-

- # Categorical expectations

- validator.expect_column_values_to_be_in_set(

- "status",

- ["pending", "confirmed", "shipped", "delivered", "cancelled"]

- )

-

- # Freshness expectation

- validator.expect_column_max_to_be_between(

- "order_date",

- min_value={"$PARAMETER": "now - timedelta(days=1)"},

- max_value={"$PARAMETER": "now"}

- )

- # Referential integrity

- validator.expect_column_values_to_be_in_set(

- "customer_id",

- value_set={"$PARAMETER": "valid_customer_ids"}

- )

- validator.save_expectation_suite(discard_failed_expectations=False)

- #### Step 3: Create Data Quality Checks with dbt

- # models/marts/schema.yml

- version: 2

-

- description: "Order fact table with data quality checks"

-

- tests:

- # Row count check

- - dbt_utils.equal_rowcount:

- compare_model: ref('stg_orders')

-

- # Freshness check

- - dbt_utils.recency:

- datepart: hour

- field: created_at

- interval: 24

-

- description: "Unique order identifier"

- tests:

- - unique

- - not_null

- - relationships:

- to: ref('dim_orders')

- field: order_id

-

- - dbt_utils.accepted_range:

- min_value: 0

- max_value: 1000000

- inclusive: true

- - dbt_expectations.expect_column_values_to_be_between:

- min_value: 0

- row_condition: "status != 'cancelled'"

-

- - name: customer_id

- tests:

- - not_null

- - relationships:

- to: ref('dim_customers')

- field: customer_id

- severity: warn

- ```

-

- #### Step 4: Implement Data Contracts

-

- ```yaml

- # contracts/orders_contract.yaml

- contract:

- name: orders_data_contract

- version: "1.0.0"

- owner: data-team@company.com

-

- schema:

- type: object

- properties:

- order_id:

- type: string

- format: uuid

- description: "Unique order identifier"

- customer_id:

- type: string

- not_null: true

- order_date:

- type: date

- not_null: true

- total_amount:

- type: decimal

- precision: 10

- scale: 2

- minimum: 0

- status:

- type: string

- enum: ["pending", "confirmed", "shipped", "delivered", "cancelled"]

-

- sla:

- freshness:

- max_delay_hours: 1

- completeness:

- min_percentage: 99.9

- accuracy:

- duplicate_tolerance: 0.01

-

- consumers:

- - name: analytics-team

- usage: "Daily reporting dashboards"

- - name: ml-team

- usage: "Churn prediction model"

- #### Step 5: Set Up Quality Monitoring Dashboard

-

- ```python

- # monitoring/quality_dashboard.py

- from datetime import datetime, timedelta

- import pandas as pd

-

- def generate_quality_report(connection, table_name: str) -> dict:

- """Generate comprehensive data quality report."""

-

- report = {

- "table": table_name,

- "timestamp": datetime.now().isoformat(),

- "checks": {}

- }

-

- # Row count check

- row_count = connection.execute(

- f"SELECT COUNT(*) FROM {table_name}"

- ).fetchone()[0]

- report["checks"]["row_count"] = {

- "value": row_count,

- "status": "pass" if row_count > 0 else "fail"

- }

-

- # Freshness check

- max_date = connection.execute(

- f"SELECT MAX(created_at) FROM {table_name}"

- ).fetchone()[0]

- hours_old = (datetime.now() - max_date).total_seconds() / 3600

- report["checks"]["freshness"] = {

- "max_timestamp": max_date.isoformat(),

- "hours_old": round(hours_old, 2),

- "status": "pass" if hours_old < 24 else "fail"

- }

-

- # Null rate check

- null_query = f"""

- SELECT

- SUM(CASE WHEN order_id IS NULL THEN 1 ELSE 0 END) as null_order_id,

- SUM(CASE WHEN customer_id IS NULL THEN 1 ELSE 0 END) as null_customer_id,

- COUNT(*) as total

- FROM {table_name}

- """

- null_result = connection.execute(null_query).fetchone()

- report["checks"]["null_rates"] = {

- "order_id": null_result[0] / null_result[2] if null_result[2] > 0 else 0,

- "customer_id": null_result[1] / null_result[2] if null_result[2] > 0 else 0,

- "status": "pass" if null_result[0] == 0 and null_result[1] == 0 else "fail"

- }

-

- # Duplicate check

- dup_query = f"""

- SELECT COUNT(*) - COUNT(DISTINCT order_id) as duplicates

- FROM {table_name}

- """

- duplicates = connection.execute(dup_query).fetchone()[0]

- report["checks"]["duplicates"] = {

- "count": duplicates,

- "status": "pass" if duplicates == 0 else "fail"

- }

-

- # Overall status

- all_passed = all(

- check["status"] == "pass"

- for check in report["checks"].values()

- )

- report["overall_status"] = "pass" if all_passed else "fail"

-

- return report

- ```

- Use this framework to choose the right approach for your data pipeline.

-

- ### Batch vs Streaming

-

- | Criteria | Batch | Streaming |

- | **Latency requirement** | Hours to days | Seconds to minutes |

- | **Data volume** | Large historical datasets | Continuous event streams |

- | **Processing complexity** | Complex transformations, ML | Simple aggregations, filtering |

- | **Cost sensitivity** | More cost-effective | Higher infrastructure cost |

- | **Error handling** | Easier to reprocess | Requires careful design |

- **Decision Tree:**

- Is real-time insight required?

- ├── Yes → Use streaming

- │ └── Is exactly-once semantics needed?

- │ ├── Yes → Kafka + Flink/Spark Structured Streaming

- │ └── No → Kafka + consumer groups

- └── No → Use batch

- └── Is data volume > 1TB daily?

- ├── Yes → Spark/Databricks

- └── No → dbt + warehouse compute

- ### Lambda vs Kappa Architecture

-

- | Aspect | Lambda | Kappa |

- |--------|--------|-------|

- | **Complexity** | Two codebases (batch + stream) | Single codebase |

- | **Maintenance** | Higher (sync batch/stream logic) | Lower |

- | **Reprocessing** | Native batch layer | Replay from source |

- | **Use case** | ML training + real-time serving | Pure event-driven |

-

- **When to choose Lambda:**

- - Need to train ML models on historical data

- - Complex batch transformations not feasible in streaming

- - Existing batch infrastructure

-

- **When to choose Kappa:**

- - Event-sourced architecture

- - All processing can be expressed as stream operations

- - Starting fresh without legacy systems

-

- ### Data Warehouse vs Data Lakehouse

-

- | **Best for** | BI, SQL analytics | ML, unstructured data |

- | **Storage cost** | Higher (proprietary format) | Lower (open formats) |

- | **Flexibility** | Schema-on-write | Schema-on-read |

- | **Performance** | Excellent for SQL | Good, improving |

- | **Ecosystem** | Mature BI tools | Growing ML tooling |

-

- ---

-

- ## Tech Stack

-

- | Category | Technologies |

- |----------|--------------|

- | **Languages** | Python, SQL, Scala |

- | **Orchestration** | Airflow, Prefect, Dagster |

- | **Transformation** | dbt, Spark, Flink |

- | **Streaming** | Kafka, Kinesis, Pub/Sub |

- | **Storage** | S3, GCS, Delta Lake, Iceberg |

- | **Warehouses** | Snowflake, BigQuery, Redshift, Databricks |

- | **Quality** | Great Expectations, dbt tests, Monte Carlo |

- | **Monitoring** | Prometheus, Grafana, Datadog |

-

- ---

-

- ## Reference Documentation

-

- ### 1. Data Pipeline Architecture

- See `references/data_pipeline_architecture.md` for:

- - Lambda vs Kappa architecture patterns

- - Batch processing with Spark and Airflow

- - Stream processing with Kafka and Flink

- - Exactly-once semantics implementation

- - Error handling and dead letter queues

-

- ### 2. Data Modeling Patterns

- See `references/data_modeling_patterns.md` for:

- - Dimensional modeling (Star/Snowflake)

- - Slowly Changing Dimensions (SCD Types 1-6)

- - Data Vault modeling

- - dbt best practices

- - Partitioning and clustering

-

- ### 3. DataOps Best Practices

- See `references/dataops_best_practices.md` for:

- - Data testing frameworks

- - Data contracts and schema validation

- - CI/CD for data pipelines

- - Observability and lineage

- - Incident response

-

- ---

-

- ## Troubleshooting

-

- ### Pipeline Failures

-

- **Symptom:** Airflow DAG fails with timeout

- ```

- Task exceeded max execution time

- ```

-

- **Solution:**

- 1. Check resource allocation

- 2. Profile slow operations

- 3. Add incremental processing

- ```python

- # Increase timeout

- default_args = {

- 'execution_timeout': timedelta(hours=2),

- }

-

- # Or use incremental loads

- WHERE updated_at > '{{ prev_ds }}'

- ```

-

- ---

-

- **Symptom:** Spark job OOM

- ```

- java.lang.OutOfMemoryError: Java heap space

- ```

-

- **Solution:**

- 1. Increase executor memory

- 2. Reduce partition size

- 3. Use disk spill

- ```python

- spark.conf.set("spark.executor.memory", "8g")

- spark.conf.set("spark.sql.shuffle.partitions", "200")

- spark.conf.set("spark.memory.fraction", "0.8")

- ```

-

- ---

-

- **Symptom:** Kafka consumer lag increasing

- ```

- Consumer lag: 1000000 messages

- ```

-

- **Solution:**

- 1. Increase consumer parallelism

- 2. Optimize processing logic

- 3. Scale consumer group

- ```bash

- # Add more partitions

- kafka-topics.sh --alter \

- --bootstrap-server localhost:9092 \

- --topic user-events \

- --partitions 24

- ```

-

- ---

-

- ### Data Quality Issues

-

- **Symptom:** Duplicate records appearing

- ```

- Expected unique, found 150 duplicates

- ```

-

- **Solution:**

- 1. Add deduplication logic

- 2. Use merge/upsert operations

- ```sql

- -- dbt incremental with dedup

- {{

- config(

- materialized='incremental',

- unique_key='order_id'

- )

- }}

-

- SELECT * FROM (

- SELECT

- *,

- ROW_NUMBER() OVER (

- PARTITION BY order_id

- ORDER BY updated_at DESC

- ) as rn

- FROM {{ source('raw', 'orders') }}

- ) WHERE rn = 1

- ```

-

- ---

-

- **Symptom:** Stale data in tables

- ```

- Last update: 3 days ago

- ```

-

- **Solution:**

- 1. Check upstream pipeline status

- 2. Verify source availability

- 3. Add freshness monitoring

- ```yaml

- # dbt freshness check

- sources:

- - name: raw

- freshness:

- warn_after: {count: 12, period: hour}

- error_after: {count: 24, period: hour}

- loaded_at_field: _loaded_at

- ```

-

- ---

-

- **Symptom:** Schema drift detected

- ```

- Column 'new_field' not in expected schema

- ```

-

- **Solution:**

- 1. Update data contract

- 2. Modify transformations

- 3. Communicate with producers

- ```python

- # Handle schema evolution

- df = spark.read.format("delta") \

- .option("mergeSchema", "true") \

- .load("/data/orders")

- ```

-

- ---

-

- ### Performance Issues

-

- **Symptom:** Query takes hours

- ```

- Query runtime: 4 hours (expected: 30 minutes)

- ```

-

- **Solution:**

- 1. Check query plan

- 2. Add proper partitioning

- 3. Optimize joins

- ```sql

- -- Before: Full table scan

- SELECT * FROM orders WHERE order_date = '2024-01-15';

-

- -- After: Partition pruning

- -- Table partitioned by order_date

- SELECT * FROM orders WHERE order_date = '2024-01-15';

-

- -- Add clustering for frequent filters

- ALTER TABLE orders CLUSTER BY (customer_id);

- ```

- **Symptom:** dbt model takes too long

- ```

- Model fct_orders completed in 45 minutes

- ```

-

- **Solution:**

- 1. Use incremental materialization

- 2. Reduce upstream dependencies

- 3. Pre-aggregate where possible

- ```sql

- -- Convert to incremental

- {{

- config(

- materialized='incremental',

- unique_key='order_id',

- on_schema_change='sync_all_columns'

- )

- }}

- SELECT * FROM {{ ref('stg_orders') }}

- {% if is_incremental() %}

- WHERE _loaded_at > (SELECT MAX(_loaded_at) FROM {{ this }})

- {% endif %}

- ```

- | Pipeline silently produces zero rows | Incremental watermark column has timezone mismatch between source and warehouse | Normalize all timestamps to UTC at extraction; add a row-count assertion (`dbt_utils.equal_rowcount`) after every incremental load |

- | Spark shuffle stage takes 10x longer than expected | Heavy data skew on the join key (a few keys hold most rows) | Salt the skewed key (`CONCAT(key, '_', MOD(RAND()*10,10))`) or use broadcast join for the smaller table |

- | Airflow scheduler marks DAG as "no tasks to run" | Circular or broken dependency reference in the DAG definition | Run `airflow dags list-import-errors` and fix the import; use `pipeline_orchestrator.py validate --dag <file> --type airflow` |

- | dbt run succeeds but downstream dashboards show stale data | Source freshness not checked before transformation begins | Add `dbt source freshness` as a prerequisite task in the DAG; define `freshness.warn_after` and `error_after` in `sources.yml` |

- | Kafka consumer lag grows unbounded | Consumer throughput is lower than producer rate; partition count too low | Increase partitions, scale consumer group, and batch `max.poll.records`; monitor with `data_quality_validator.py profile` on output tables |

- | Data quality validator reports false-positive anomalies | Default z-score threshold (3.0) is too tight for heavy-tailed distributions | Raise the z-threshold or switch to IQR mode with a higher multiplier; re-run `data_quality_validator.py validate --detect-anomalies` |

- | Cost estimate differs significantly from actual cloud bill | The tool uses heuristic estimates without live warehouse metadata | Treat `etl_performance_optimizer.py estimate-cost` output as a directional guide; cross-reference with the warehouse query history view |

-

- ---

-

- ## Success Criteria

-

- - **Pipeline SLA above 99.5%** -- fewer than 2 unplanned failures per month across all production DAGs.

- - **Data freshness under 15 minutes** for streaming pipelines and under 2 hours for batch pipelines measured at the mart layer.

- - **Data quality score >= 95%** on completeness, uniqueness, validity, consistency, and accuracy (as reported by `data_quality_validator.py`).

- - **Zero duplicate records** in all fact and dimension tables enforced by primary key merge/upsert logic and dbt uniqueness tests.

- - **Query optimization savings >= 30%** in compute cost or execution time after applying recommendations from `etl_performance_optimizer.py analyze-sql`.

- - **Schema drift detected within one pipeline run** -- all contract violations surfaced automatically before data reaches the mart layer.

- - **Incident MTTR under 30 minutes** for P1 pipeline failures with documented runbooks referencing the troubleshooting table above.

- ## Scope & Limitations

-

- **This skill covers:**

- - End-to-end batch and streaming pipeline design (Airflow, Prefect, Dagster, Kafka, Spark)

- - Data quality validation, profiling, anomaly detection, and Great Expectations suite generation

- - SQL and Spark performance analysis with actionable optimization recommendations

- - Data modeling patterns (star schema, snowflake, data vault, SCD types)

- **This skill does NOT cover:**

- - Machine learning model training and serving -- see `senior-ml-engineer`

- - Statistical experiment design and hypothesis testing -- see `senior-data-scientist`

- - Cloud infrastructure provisioning (Terraform, CloudFormation) -- see `senior-devops` or `aws-solution-architect`

- - Application-level security hardening and vulnerability scanning -- see `senior-secops` or `senior-security`

- | Skill | Integration | Data Flow |

- |-------|-------------|-----------|

- | `senior-data-scientist` | Feature engineering pipelines consume cleaned data from the mart layer | Data engineer publishes curated datasets; data scientist runs experiments and feeds feature definitions back for productionization |

- | `senior-ml-engineer` | ML training pipelines depend on feature store tables built by data engineering | Data engineer maintains feature store refresh; ML engineer deploys model artifacts and monitoring |

- | `senior-devops` | CI/CD for dbt projects, Airflow deployment, and container orchestration | Data engineer defines pipeline code; DevOps manages infrastructure, Docker images, and deployment workflows |

- | `senior-architect` | Architecture reviews for data platform decisions (lakehouse vs warehouse, Lambda vs Kappa) | Data engineer proposes designs; architect validates against enterprise standards and non-functional requirements |

- | `senior-backend` | API-sourced ingestion and event-driven pipelines consume backend service events | Backend publishes events to Kafka/queues; data engineer builds consumers and transformation layers |

- | `code-reviewer` | Pipeline code reviews for Airflow DAGs, dbt models, and Spark jobs | Data engineer submits PRs; code reviewer validates SQL patterns, idempotency, and error handling |

-

- ---

-

- ## Tool Reference

-

- ### pipeline_orchestrator.py

-

- **Purpose:** Generate pipeline configurations for Airflow, Prefect, and Dagster. Supports ETL pattern generation, dependency management, scheduling, and DAG validation.

-

- **Subcommands:**

-

- #### `generate` -- Generate pipeline code

-

- ```bash

- python scripts/pipeline_orchestrator.py generate \

- --type airflow \

- --source postgres \

- --destination snowflake \

- --tables orders,customers \

- --schedule "0 5 * * *" \

- --mode incremental \

- --output dags/my_pipeline.py

- ```

-

- | Flag | Short | Required | Default | Description |

- |------|-------|----------|---------|-------------|

- | `--type` | `-t` | Yes | -- | Pipeline framework: `airflow`, `prefect`, `dagster` |

- | `--source` | `-s` | No | `postgres` | Source system type |

- | `--destination` | `-d` | No | `snowflake` | Destination system type |

- | `--tables` | -- | No | -- | Comma-separated list of tables to extract |

- | `--config` | `-c` | No | -- | Configuration YAML file (overrides other source/dest flags) |

- | `--output` | `-o` | No | stdout | Output file path for generated code |

- | `--name` | `-n` | No | auto-generated | Pipeline name |

- | `--schedule` | -- | No | `0 5 * * *` | Cron schedule expression |

- | `--mode` | -- | No | `incremental` | Load mode: `incremental` or `full` |

-

- **Output formats:** Generated Python code written to file or printed to stdout.

-

- #### `validate` -- Validate existing pipeline code

-

- ```bash

- python scripts/pipeline_orchestrator.py validate \

- --dag dags/my_dag.py \

- --type airflow

- ```

-

- | Flag | Short | Required | Default | Description |

- |------|-------|----------|---------|-------------|

- | `--dag` | -- | Yes | -- | Path to the DAG/flow file to validate |

- | `--type` | `-t` | Yes | -- | Framework type: `airflow`, `prefect`, `dagster` |

-

- **Output formats:** JSON validation result with `valid` (boolean), `issues` (list), and `warnings` (list). Exits with code 0 on success, 1 on failure.

-

- #### `template` -- Generate from ETL pattern template

-

- ```bash

- python scripts/pipeline_orchestrator.py template \

- --pattern extract-load \

- --type airflow \

- --source postgres \

- --destination snowflake \

- --tables orders,customers \

- --output dags/el_pipeline.py

- ```

-

- | Flag | Short | Required | Default | Description |

- |------|-------|----------|---------|-------------|

- | `--pattern` | `-p` | Yes | -- | ETL pattern: `extract-load`, `transform`, `cdc` |

- | `--type` | `-t` | Yes | -- | Framework type: `airflow`, `prefect`, `dagster` |

- | `--source` | `-s` | Yes | -- | Source system type |

- | `--destination` | `-d` | Yes | -- | Destination system type |

- | `--tables` | -- | Yes | -- | Comma-separated list of tables |

- | `--output` | `-o` | No | stdout | Output file path |

-

- ---

-

- ### data_quality_validator.py

-

- **Purpose:** Comprehensive data quality validation including schema checking, data profiling, anomaly detection, Great Expectations suite generation, and data contract enforcement. Supports CSV, JSON, and JSONL inputs.

-

- **Global flags:**

-

- | Flag | Short | Description |

- |------|-------|-------------|

- | `--verbose` | `-v` | Enable verbose logging output |

-

- **Subcommands:**

-

- #### `validate` -- Validate data against a schema

-

- ```bash

- python scripts/data_quality_validator.py validate data.csv \

- --schema schema.json \

- --detect-anomalies \

- --output report.json \

- --json

- ```

-

- | Flag | Short | Required | Default | Description |

- |------|-------|----------|---------|-------------|

- | `input` (positional) | -- | Yes | -- | Input data file (CSV, JSON, JSONL) |

- | `--schema` | `-s` | No | -- | Schema file (JSON) to validate against |

- | `--output` | `-o` | No | stdout | Output report file path |

- | `--json` | -- | No | `false` | Output as JSON instead of human-readable text |

- | `--detect-anomalies` | -- | No | `false` | Also run statistical anomaly detection (z-score and IQR) |

-

- **Output formats:** Human-readable validation report (default) or JSON with per-check results, severity, failed rows, and overall quality score.

-

- #### `profile` -- Generate a statistical data profile

-

- ```bash

- python scripts/data_quality_validator.py profile data.csv \

- --output profile.json \

- --json

- ```

-

- | Flag | Short | Required | Default | Description |

- |------|-------|----------|---------|-------------|

- | `input` (positional) | -- | Yes | -- | Input data file |

- | `--output` | `-o` | No | stdout | Output profile file path |

- | `--json` | -- | No | `false` | Output as JSON |

-

- **Output formats:** Per-column statistics including null counts, unique counts, min/max/mean/median/std_dev for numerics, length stats for strings, top values, and detected patterns.

-

- #### `generate-suite` -- Generate a Great Expectations suite

-

- ```bash

- python scripts/data_quality_validator.py generate-suite data.csv \

- --output expectations.json

- ```

-

- | Flag | Short | Required | Default | Description |

- |------|-------|----------|---------|-------------|

- | `input` (positional) | -- | Yes | -- | Input data file to base expectations on |

- | `--output` | `-o` | No | stdout | Output expectations JSON file |

-

- **Output formats:** JSON expectation suite compatible with Great Expectations, derived from the data profile.

-

- #### `contract` -- Validate against a data contract

-

- ```bash

- python scripts/data_quality_validator.py contract data.csv \

- --contract contract.yaml \

- --output report.json \

- --json

- ```

-

- | Flag | Short | Required | Default | Description |

- |------|-------|----------|---------|-------------|

- | `input` (positional) | -- | Yes | -- | Input data file |

- | `--contract` | `-c` | Yes | -- | Data contract file (YAML or JSON) |

- | `--output` | `-o` | No | stdout | Output report file path |

- | `--json` | -- | No | `false` | Output as JSON |

-

- **Output formats:** Contract validation report showing SLA compliance (freshness, completeness, accuracy) and per-field results.

-

- #### `schema` -- Infer and generate a schema from data

-

- ```bash

- python scripts/data_quality_validator.py schema data.csv \

- --output schema.json

- ```

-

- | Flag | Short | Required | Default | Description |

- |------|-------|----------|---------|-------------|

- | `input` (positional) | -- | Yes | -- | Input data file |

- | `--output` | `-o` | No | stdout | Output schema JSON file |

-

- **Output formats:** JSON schema with inferred column types, nullability, uniqueness, and detected patterns.

- ### etl_performance_optimizer.py

-

- **Purpose:** ETL/ELT performance analysis and optimization. Analyzes SQL queries, Spark job metrics, partition strategies, and estimates cloud warehouse costs. Provides actionable recommendations sorted by priority and severity.

-

- **Global flags:**

-

- | Flag | Short | Description |

- |------|-------|-------------|

- | `--verbose` | `-v` | Enable verbose logging output |

-

- **Subcommands:**

-

- #### `analyze-sql` -- Analyze a SQL query for optimization

-

- ```bash

- python scripts/etl_performance_optimizer.py analyze-sql query.sql \

- --warehouse snowflake \

- --stats data_stats.json \

- --output recommendations.json \

- --json

- ```

-

- | Flag | Short | Required | Default | Description |

- |------|-------|----------|---------|-------------|

- | `input` (positional) | -- | Yes | -- | SQL file path or inline query string |

- | `--warehouse` | `-w` | No | -- | Target warehouse: `bigquery`, `snowflake`, `redshift`, `databricks` |

- | `--stats` | `-s` | No | -- | Data statistics JSON file for context-aware recommendations |

- | `--output` | `-o` | No | stdout | Output file path |

- | `--json` | -- | No | `false` | Output as JSON |

-

- **Output formats:** Prioritized list of recommendations with category, severity, title, description, current issue, recommendation, expected improvement, and implementation steps.

-

- #### `analyze-spark` -- Analyze Spark job metrics

-

- ```bash

- python scripts/etl_performance_optimizer.py analyze-spark spark_metrics.json \

- --output report.json \

- --json

- ```

-

- | Flag | Short | Required | Default | Description |

- |------|-------|----------|---------|-------------|

- | `input` (positional) | -- | Yes | -- | Spark metrics JSON file (from Spark History Server or custom export) |

- | `--output` | `-o` | No | stdout | Output file path |

- | `--json` | -- | No | `false` | Output as JSON |

-

- **Output formats:** Analysis of shuffle, memory, GC pressure, skew ratio, and task failure rates with targeted recommendations.

-

- #### `optimize-partition` -- Recommend partition strategies

-

- ```bash

- python scripts/etl_performance_optimizer.py optimize-partition data_stats.json \

- --output partitions.json \

- --json

- ```

-

- | Flag | Short | Required | Default | Description |

- |------|-------|----------|---------|-------------|

- | `input` (positional) | -- | Yes | -- | Data statistics JSON file with column cardinality and distribution info |

- | `--output` | `-o` | No | stdout | Output file path |

- | `--json` | -- | No | `false` | Output as JSON |

-

- **Output formats:** Partition strategy per column including type (range, hash, list), recommended partition count, target partition size in MB, reasoning, and implementation SQL.

-

- #### `estimate-cost` -- Estimate query execution cost

-

- ```bash

- python scripts/etl_performance_optimizer.py estimate-cost query.sql \

- --warehouse snowflake \

- --stats data_stats.json \

- --output cost.json \

- --json

- ```

-

- | Flag | Short | Required | Default | Description |

- |------|-------|----------|---------|-------------|

- | `input` (positional) | -- | Yes | -- | SQL file path or inline query string |

- | `--warehouse` | `-w` | Yes | -- | Target warehouse: `bigquery`, `snowflake`, `redshift`, `databricks` |

- | `--stats` | `-s` | No | -- | Data statistics JSON file for more accurate estimates |

- | `--output` | `-o` | No | stdout | Output file path |

- | `--json` | -- | No | `false` | Output as JSON |

-

- **Output formats:** Cost breakdown with compute, storage, and data transfer costs in USD plus underlying assumptions.

-

- #### `template` -- Generate template files for input

-

- ```bash

- python scripts/etl_performance_optimizer.py template data_stats --output stats_template.json

- python scripts/etl_performance_optimizer.py template spark_metrics --output metrics_template.json

- ```

-

- | Flag | Short | Required | Default | Description |

- |------|-------|----------|---------|-------------|

- | `template` (positional) | -- | Yes | -- | Template type: `data_stats` or `spark_metrics` |

- | `--output` | `-o` | No | stdout | Output file path |

-

- **Output formats:** JSON template with placeholder values showing the expected structure for `--stats` input files or Spark metrics files.