Data Pipeline Engineer

# Airflow DAG Reference
# Complete ETL pipeline with sensors, task groups, and quality checks

from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.providers.apache.spark.operators.spark_submit import SparkSubmitOperator
from airflow.providers.dbt.cloud.operators.dbt import DbtCloudRunJobOperator
from airflow.sensors.external_task import ExternalTaskSensor
from airflow.utils.task_group import TaskGroup

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

with DAG(
    dag_id='etl_orders_pipeline',
    default_args=default_args,
    description='Daily orders ETL pipeline',
    schedule_interval='0 6 * * *',  # 6 AM daily
    start_date=datetime(2024, 1, 1),
    catchup=False,
    max_active_runs=1,
    tags=['etl', 'orders', 'production'],
) as dag:

    # Wait for upstream data
    wait_for_source = ExternalTaskSensor(
        task_id='wait_for_source_data',
        external_dag_id='source_system_export',
        external_task_id='export_complete',
        timeout=3600,
        poke_interval=60,
        mode='reschedule',
    )

    # Extract and load to bronze
    with TaskGroup(group_id='extract_load') as extract_load:
        extract_orders = SparkSubmitOperator(
            task_id='extract_orders',
            application='/opt/spark/jobs/extract_orders.py',
            conn_id='spark_default',
            conf={
                'spark.sql.shuffle.partitions': '200',
                'spark.dynamicAllocation.enabled': 'true',
            },
            application_args=[
                '--date', '{{ ds }}',
                '--source', 's3://source-bucket/orders/',
                '--target', 's3://bronze-bucket/orders/',
            ],
        )

        extract_customers = SparkSubmitOperator(
            task_id='extract_customers',
            application='/opt/spark/jobs/extract_customers.py',
            conn_id='spark_default',
            application_args=[
                '--date', '{{ ds }}',
                '--source', 's3://source-bucket/customers/',
                '--target', 's3://bronze-bucket/customers/',
            ],
        )

    # Data quality checks on bronze
    def run_bronze_quality_checks(**context):
        from great_expectations.data_context import DataContext

        ge_context = DataContext('/opt/great_expectations')
        result = ge_context.run_checkpoint(
            checkpoint_name='bronze_orders_checkpoint',
            batch_request={
                'datasource_name': 's3_bronze',
                'data_asset_name': 'orders',
                'data_connector_query': {
                    'batch_filter_parameters': {
                        'date': context['ds']
                    }
                }
            }
        )
        if not result.success:
            raise ValueError('Bronze data quality checks failed')

    bronze_quality = PythonOperator(
        task_id='bronze_quality_checks',
        python_callable=run_bronze_quality_checks,
    )

    # Transform with dbt (bronze → silver → gold)
    dbt_run = DbtCloudRunJobOperator(
        task_id='dbt_transform',
        job_id=12345,
        check_interval=30,
        timeout=7200,
        wait_for_termination=True,
    )

    # Gold layer quality checks
    gold_quality = PythonOperator(
        task_id='gold_quality_checks',
        python_callable=lambda: print('Running gold quality checks'),
    )

    # Notify downstream
    def notify_completion(**context):
        # Send Slack notification, update data catalog, etc.
        pass

    notify = PythonOperator(
        task_id='notify_completion',
        python_callable=notify_completion,
    )

    # DAG dependencies
    wait_for_source >> extract_load >> bronze_quality >> dbt_run >> gold_quality >> notify

{
  "expectation_suite_name": "orders_bronze",
  "meta": {
    "great_expectations_version": "0.18.0"
  },
  "expectations": [
    {
      "expectation_type": "expect_table_row_count_to_be_between",
      "kwargs": {
        "min_value": 1000,
        "max_value": 10000000
      },
      "meta": {
        "notes": "Alert if unusually low/high volume"
      }
    },
    {
      "expectation_type": "expect_column_values_to_not_be_null",
      "kwargs": {
        "column": "order_id"
      }
    },
    {
      "expectation_type": "expect_column_values_to_be_unique",
      "kwargs": {
        "column": "order_id"
      }
    },
    {
      "expectation_type": "expect_column_values_to_not_be_null",
      "kwargs": {
        "column": "customer_id"
      }
    },
    {
      "expectation_type": "expect_column_values_to_be_between",
      "kwargs": {
        "column": "total_amount",
        "min_value": 0,
        "max_value": 100000
      },
      "meta": {
        "notes": "Flag outlier orders for review"
      }
    },
    {
      "expectation_type": "expect_column_values_to_match_regex",
      "kwargs": {
        "column": "email",
        "regex": "^[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\\.[a-zA-Z0-9-.]+$"
      }
    },
    {
      "expectation_type": "expect_column_pair_values_a_to_be_greater_than_b",
      "kwargs": {
        "column_A": "updated_at",
        "column_B": "created_at",
        "or_equal": true
      }
    },
    {
      "expectation_type": "expect_column_values_to_be_in_set",
      "kwargs": {
        "column": "status",
        "value_set": ["pending", "processing", "shipped", "delivered", "cancelled", "refunded"]
      }
    },
    {
      "expectation_type": "expect_column_values_to_be_dateutil_parseable",
      "kwargs": {
        "column": "order_date"
      }
    },
    {
      "expectation_type": "expect_column_values_to_be_of_type",
      "kwargs": {
        "column": "quantity",
        "type_": "int"
      }
    },
    {
      "expectation_type": "expect_column_values_to_be_between",
      "kwargs": {
        "column": "quantity",
        "min_value": 1,
        "max_value": 1000
      }
    },
    {
      "expectation_type": "expect_table_columns_to_match_set",
      "kwargs": {
        "column_set": [
          "order_id",
          "customer_id",
          "order_date",
          "total_amount",
          "status",
          "email",
          "quantity",
          "created_at",
          "updated_at",
          "_loaded_at"
        ]
      }
    }
  ]
}

# Spark Streaming Reference
# Complete Kafka-to-Delta Lake streaming processor with windowing

from pyspark.sql import SparkSession
from pyspark.sql.functions import (
    from_json, col, window, sum as spark_sum,
    count, avg, expr, current_timestamp
)
from pyspark.sql.types import (
    StructType, StructField, StringType,
    DoubleType, TimestampType, IntegerType
)

spark = SparkSession.builder \
    .appName("OrderStreamProcessor") \
    .config("spark.sql.streaming.checkpointLocation", "s3://checkpoints/orders") \
    .getOrCreate()

# Define schema for incoming events
order_schema = StructType([
    StructField("order_id", StringType(), False),
    StructField("customer_id", StringType(), False),
    StructField("product_id", StringType(), False),
    StructField("quantity", IntegerType(), False),
    StructField("unit_price", DoubleType(), False),
    StructField("event_time", TimestampType(), False),
])

# Read from Kafka
orders_stream = spark \
    .readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "kafka:9092") \
    .option("subscribe", "orders") \
    .option("startingOffsets", "latest") \
    .option("failOnDataLoss", "false") \
    .load()

# Parse JSON and apply schema
parsed_orders = orders_stream \
    .select(from_json(col("value").cast("string"), order_schema).alias("data")) \
    .select("data.*") \
    .withColumn("total_amount", col("quantity") * col("unit_price")) \
    .withWatermark("event_time", "10 minutes")

# Aggregate by 5-minute windows
windowed_metrics = parsed_orders \
    .groupBy(
        window(col("event_time"), "5 minutes", "1 minute"),
        col("product_id")
    ) \
    .agg(
        count("order_id").alias("order_count"),
        spark_sum("quantity").alias("total_quantity"),
        spark_sum("total_amount").alias("total_revenue"),
        avg("total_amount").alias("avg_order_value")
    ) \
    .select(
        col("window.start").alias("window_start"),
        col("window.end").alias("window_end"),
        "product_id",
        "order_count",
        "total_quantity",
        "total_revenue",
        "avg_order_value",
        current_timestamp().alias("processed_at")
    )

# Write to Delta Lake
query = windowed_metrics \
    .writeStream \
    .format("delta") \
    .outputMode("append") \
    .option("checkpointLocation", "s3://checkpoints/metrics") \
    .option("mergeSchema", "true") \
    .trigger(processingTime="30 seconds") \
    .start("s3://gold-bucket/order_metrics")

# Also write to Kafka for real-time dashboards
kafka_query = windowed_metrics \
    .selectExpr("to_json(struct(*)) as value") \
    .writeStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "kafka:9092") \
    .option("topic", "order_metrics") \
    .option("checkpointLocation", "s3://checkpoints/kafka-metrics") \
    .start()

spark.streams.awaitAnyTermination()

#!/bin/bash
# Data Pipeline Engineer Skill Validation Script
# Validates data pipeline configurations for best practices

set -e

ERRORS=0
WARNINGS=0

echo "═══════════════════════════════════════════════════════════════"
echo "Data Pipeline Engineer Skill Validator"
echo "═══════════════════════════════════════════════════════════════"
echo ""

# Check dbt project
check_dbt() {
    echo "📊 Checking dbt project..."

    if [ -f "dbt_project.yml" ]; then
        echo "  Found dbt_project.yml"

        # Check for version
        if ! grep -q "version:" dbt_project.yml 2>/dev/null; then
            echo "⚠️  WARN: dbt_project.yml missing version"
            ((WARNINGS++))
        fi

        # Check for models directory
        if [ ! -d "models" ]; then
            echo "❌ ERROR: models/ directory not found"
            ((ERRORS++))
        fi

        # Check for staging layer
        if [ ! -d "models/staging" ]; then
            echo "⚠️  WARN: models/staging/ not found (recommended for medallion architecture)"
            ((WARNINGS++))
        fi

        # Check for marts layer
        if [ ! -d "models/marts" ]; then
            echo "⚠️  WARN: models/marts/ not found (recommended for gold layer)"
            ((WARNINGS++))
        fi

        # Check models for incremental config
        for model in models/**/*.sql; do
            [ -f "$model" ] || continue

            # Check if large tables are incremental
            if grep -q "materialized='table'" "$model" 2>/dev/null; then
                if ! grep -q "is_incremental" "$model" 2>/dev/null; then
                    echo "⚠️  WARN: $model uses table materialization but not incremental"
                    ((WARNINGS++))
                fi
            fi

            # Check for hardcoded dates
            if grep -qE "'202[0-9]-[0-9]{2}-[0-9]{2}'" "$model" 2>/dev/null; then
                echo "⚠️  WARN: $model contains hardcoded dates"
                ((WARNINGS++))
            fi
        done

        # Check for schema YAML files
        schema_files=$(find models -name "*.yml" 2>/dev/null | wc -l)
        if [ "$schema_files" -eq 0 ]; then
            echo "⚠️  WARN: No schema YAML files found (add tests and documentation)"
            ((WARNINGS++))
        fi
    else
        echo "ℹ️  No dbt project found"
    fi
}

# Check Airflow DAGs
check_airflow() {
    echo ""
    echo "🔄 Checking Airflow DAGs..."

    for dag in dags/*.py; do
        [ -f "$dag" ] || continue

        echo "  Checking: $dag"

        # Check for retries
        if ! grep -q "retries" "$dag" 2>/dev/null; then
            echo "⚠️  WARN: $dag missing retry configuration"
            ((WARNINGS++))
        fi

        # Check for catchup=False (usually desired)
        if grep -q "catchup=True" "$dag" 2>/dev/null; then
            echo "ℹ️  INFO: $dag has catchup=True (ensure this is intentional)"
        fi

        # Check for max_active_runs
        if ! grep -q "max_active_runs" "$dag" 2>/dev/null; then
            echo "⚠️  WARN: $dag missing max_active_runs (could cause parallel run issues)"
            ((WARNINGS++))
        fi

        # Check for email_on_failure
        if ! grep -q "email_on_failure" "$dag" 2>/dev/null; then
            echo "⚠️  WARN: $dag missing email_on_failure notification"
            ((WARNINGS++))
        fi

        # Check for hardcoded connections
        if grep -qE "host\s*=\s*['\"]" "$dag" 2>/dev/null; then
            echo "❌ ERROR: $dag contains hardcoded host (use Airflow connections)"
            ((ERRORS++))
        fi
    done
}

# Check Spark jobs
check_spark() {
    echo ""
    echo "⚡ Checking Spark jobs..."

    for spark_job in *.py spark/*.py jobs/*.py; do
        [ -f "$spark_job" ] || continue

        # Check if it's a Spark file
        if ! grep -q "SparkSession\|pyspark" "$spark_job" 2>/dev/null; then
            continue
        fi

        echo "  Checking: $spark_job"

        # Check for checkpoint location in streaming
        if grep -q "readStream\|writeStream" "$spark_job" 2>/dev/null; then
            if ! grep -q "checkpointLocation" "$spark_job" 2>/dev/null; then
                echo "❌ ERROR: $spark_job streaming job missing checkpointLocation"
                ((ERRORS++))
            fi
        fi

        # Check for watermark in streaming
        if grep -q "readStream" "$spark_job" 2>/dev/null; then
            if ! grep -q "withWatermark" "$spark_job" 2>/dev/null; then
                echo "⚠️  WARN: $spark_job streaming job missing watermark (may accumulate state)"
                ((WARNINGS++))
            fi
        fi

        # Check for shuffle partition tuning
        if grep -q "groupBy\|join" "$spark_job" 2>/dev/null; then
            if ! grep -q "shuffle.partitions" "$spark_job" 2>/dev/null; then
                echo "ℹ️  INFO: $spark_job may benefit from tuning spark.sql.shuffle.partitions"
            fi
        fi
    done
}

# Check data quality
check_data_quality() {
    echo ""
    echo "✅ Checking data quality setup..."

    # Check for Great Expectations
    if [ -d "great_expectations" ]; then
        echo "  Found Great Expectations configuration"

        if [ ! -d "great_expectations/expectations" ]; then
            echo "⚠️  WARN: No expectation suites found"
            ((WARNINGS++))
        fi

        if [ ! -d "great_expectations/checkpoints" ]; then
            echo "⚠️  WARN: No checkpoints configured"
            ((WARNINGS++))
        fi
    fi

    # Check for dbt tests
    if [ -f "dbt_project.yml" ]; then
        test_count=$(grep -r "tests:" models/ 2>/dev/null | wc -l)
        if [ "$test_count" -lt 5 ]; then
            echo "⚠️  WARN: Few dbt tests found (add schema tests)"
            ((WARNINGS++))
        fi
    fi
}

# Run all checks
check_dbt
check_airflow
check_spark
check_data_quality

# Summary
echo ""
echo "═══════════════════════════════════════════════════════════════"
echo "Validation Complete"
echo "═══════════════════════════════════════════════════════════════"
echo "Errors:   $ERRORS"
echo "Warnings: $WARNINGS"
echo ""

if [ $ERRORS -gt 0 ]; then
    echo "❌ Validation FAILED - fix errors before deployment"
    exit 1
elif [ $WARNINGS -gt 5 ]; then
    echo "⚠️  Validation PASSED with warnings - review recommended"
    exit 0
else
    echo "✅ Validation PASSED"
    exit 0
fi