┌─────────────┐     ┌─────────────┐     ┌─────────────┐
│   Sources   │────▶│  Transform  │────▶│   Target    │
│  (S3, API)  │     │  (Spark)    │     │  (Snowflake)│
└─────────────┘     └─────────────┘     └─────────────┘
       │                   │                   │
       │                   ▼                   │
       │           ┌─────────────┐             │
       └──────────▶│   Quality   │◀────────────┘
                   │   Checks    │
                   └─────────────┘

┌─────────────┐     ┌─────────────┐     ┌─────────────┐
│   Kafka     │────▶│  Processor  │────▶│  Real-time  │
│   Topics    │     │  (Flink)    │     │    Store    │
└─────────────┘     └─────────────┘     └─────────────┘
                           │
                           ▼
                    ┌─────────────┐
                    │   Batch     │
                    │   Layer     │
                    └─────────────┘

from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.providers.snowflake.operators.snowflake import SnowflakeOperator
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(
    dag_id='daily_etl_pipeline',
    default_args=default_args,
    description='Daily ETL for analytics',
    schedule_interval='0 6 * * *',
    start_date=days_ago(1),
    catchup=False,
    tags=['etl', 'daily'],
) as dag:

    extract_task = PythonOperator(
        task_id='extract_data',
        python_callable=extract_from_source,
        op_kwargs={'date': '{{ ds }}'},
    )

    transform_task = PythonOperator(
        task_id='transform_data',
        python_callable=transform_data,
    )

    quality_check = PythonOperator(
        task_id='quality_check',
        python_callable=run_quality_checks,
    )

    load_task = SnowflakeOperator(
        task_id='load_to_snowflake',
        sql='sql/load_data.sql',
        snowflake_conn_id='snowflake_default',
        params={'date': '{{ ds }}'},
    )

    extract_task >> transform_task >> quality_check >> load_task

from airflow.decorators import dag, task
from datetime import datetime

@dag(
    schedule_interval='@daily',
    start_date=datetime(2024, 1, 1),
    catchup=False,
)
def dynamic_etl():

    @task
    def get_sources():
        return ['source_a', 'source_b', 'source_c']

    @task
    def extract(source: str):
        print(f'Extracting from {source}')
        return f'{source}_data'

    @task
    def transform(data: str):
        print(f'Transforming {data}')
        return f'{data}_transformed'

    @task
    def load(data: list):
        print(f'Loading {len(data)} datasets')

    sources = get_sources()
    extracted = extract.expand(source=sources)
    transformed = transform.expand(data=extracted)
    load(transformed)

dag = dynamic_etl()

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, when, lit, current_timestamp
from pyspark.sql.types import StructType, StructField, StringType, IntegerType

spark = SparkSession.builder \
    .appName("ETL Pipeline") \
    .config("spark.sql.adaptive.enabled", "true") \
    .config("spark.sql.shuffle.partitions", "200") \
    .getOrCreate()


def read_source_data(path: str, date: str):
    return spark.read \
        .option("header", "true") \
        .option("inferSchema", "true") \
        .csv(f"{path}/date={date}/*.csv")


def transform_data(df):
    return df \
        .withColumn("amount_clean",
            when(col("amount").isNull(), 0).otherwise(col("amount"))) \
        .withColumn("category_normalized",
            when(col("category").isin(["A", "B"]), col("category"))
            .otherwise("OTHER")) \
        .withColumn("processed_at", current_timestamp()) \
        .filter(col("status") == "active") \
        .dropDuplicates(["id"])


def aggregate_metrics(df):
    return df.groupBy("date", "category") \
        .agg(
            F.count("*").alias("total_count"),
            F.sum("amount").alias("total_amount"),
            F.avg("amount").alias("avg_amount"),
            F.countDistinct("user_id").alias("unique_users")
        )


def write_to_warehouse(df, table_name: str, partition_cols: list):
    df.write \
        .mode("overwrite") \
        .partitionBy(partition_cols) \
        .format("delta") \
        .saveAsTable(table_name)

undefined

undefined

undefined

-- Fact Table
CREATE TABLE fact_sales (
    sale_id BIGINT PRIMARY KEY,
    date_key INT REFERENCES dim_date(date_key),
    product_key INT REFERENCES dim_product(product_key),
    customer_key INT REFERENCES dim_customer(customer_key),
    store_key INT REFERENCES dim_store(store_key),
    quantity INT,
    unit_price DECIMAL(10,2),
    total_amount DECIMAL(10,2),
    discount_amount DECIMAL(10,2)
);

-- Dimension Tables
CREATE TABLE dim_date (
    date_key INT PRIMARY KEY,
    full_date DATE,
    year INT,
    quarter INT,
    month INT,
    week INT,
    day_of_week INT,
    is_weekend BOOLEAN,
    is_holiday BOOLEAN
);

CREATE TABLE dim_product (
    product_key INT PRIMARY KEY,
    product_id VARCHAR(50),
    product_name VARCHAR(200),
    category VARCHAR(100),
    subcategory VARCHAR(100),
    brand VARCHAR(100),
    -- SCD Type 2 columns
    effective_date DATE,
    end_date DATE,
    is_current BOOLEAN
);

{{ config(materialized='view') }}

with source as (
    select * from {{ source('raw', 'orders') }}
),

renamed as (
    select
        order_id,
        customer_id,
        order_date,
        cast(total_amount as decimal(10,2)) as total_amount,
        status,
        _loaded_at
    from source
    where order_date >= '2023-01-01'
)

select * from renamed

{{ config(
    materialized='incremental',
    unique_key='order_id',
    incremental_strategy='merge'
) }}

with orders as (
    select * from {{ ref('stg_orders') }}
),

customers as (
    select * from {{ ref('dim_customers') }}
),

final as (
    select
        o.order_id,
        o.customer_id,
        c.customer_segment,
        o.order_date,
        o.total_amount,
        o.status,
        current_timestamp() as processed_at
    from orders o
    left join customers c on o.customer_id = c.customer_id
    {% if is_incremental() %}
    where o._loaded_at > (select max(processed_at) from {{ this }})
    {% endif %}
)

select * from final

import great_expectations as gx

context = gx.get_context()

undefined

from dataclasses import dataclass
from typing import Callable

@dataclass
class QualityCheck:
    name: str
    check_fn: Callable
    severity: str  # 'error' or 'warning'


def check_no_nulls(df, column: str) -> bool:
    null_count = df.filter(col(column).isNull()).count()
    return null_count == 0


def check_referential_integrity(df, ref_df, key: str) -> bool:
    orphans = df.join(ref_df, key, "left_anti")
    return orphans.count() == 0


def check_freshness(df, timestamp_col: str, max_hours: int) -> bool:
    from pyspark.sql.functions import max as spark_max, current_timestamp
    max_ts = df.agg(spark_max(timestamp_col)).collect()[0][0]
    hours_old = (datetime.now() - max_ts).total_seconds() / 3600
    return hours_old <= max_hours


def run_quality_checks(df, checks: list[QualityCheck]):
    results = []
    for check in checks:
        passed = check.check_fn(df)
        results.append({
            'check': check.name,
            'passed': passed,
            'severity': check.severity
        })

    errors = [r for r in results if not r['passed'] and r['severity'] == 'error']
    if errors:
        raise DataQualityError(f"Quality checks failed: {errors}")

    return results

from pyflink.datastream import StreamExecutionEnvironment
from pyflink.table import StreamTableEnvironment

env = StreamExecutionEnvironment.get_execution_environment()
t_env = StreamTableEnvironment.create(env)

undefined

undefined

undefined