Building Robust ETL/ELT Pipelines

graph TD
    A[Data Sources] -->|Extract| B{ETL/ELT Engine}
    B -->|Transform| C[Data Warehouse]
    B -->|Load| D[Data Lake]
    C --> E[Analytics]
    D --> F[ML Models]
    E --> G[Dashboards]
    F --> H[Applications]

from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.providers.amazon.aws.hooks.s3 import S3Hook
from airflow.providers.postgres.hooks.postgres import PostgresHook
from airflow.providers.snowflake.operators.snowflake import SnowflakeOperator

def extract_data():
    """Extract data from source systems."""
    # Example: Extract from API
    import requests
    import pandas as pd

    response = requests.get("https://api.example.com/data")
    data = response.json()
    df = pd.DataFrame(data)

    # Save to S3
    s3_hook = S3Hook(aws_conn_id='aws_default')
    s3_hook.load_string(
        string_data=df.to_csv(index=False),
        key='raw/data.csv',
        bucket_name='my-data-lake',
        replace=True
    )

def transform_data():
    """Transform data using Spark."""
    from pyspark.sql import SparkSession
    from pyspark.sql.functions import col, to_date

    spark = SparkSession.builder \
        .appName("ETL-Transform") \
        .getOrCreate()

    # Read from S3
    df = spark.read.csv("s3a://my-data-lake/raw/", header=True)

    # Perform transformations
    transformed_df = (df
        .withColumn("date", to_date(col("timestamp")))
        .withColumn("amount", col("amount").cast("double"))
        .filter(col("status") == "completed")
    )

    # Write to processed zone
    (transformed_df.write
        .mode("overwrite")
        .parquet("s3a://my-data-lake/processed/")
    )

def load_to_warehouse():
    """Load processed data into Snowflake."""
    # This would be implemented as a SnowflakeOperator
    pass

# Define the DAG
default_args = {
    'owner': 'data_engineering',
    'depends_on_past': False,
    'start_date': datetime(2023, 1, 1),
    'email_on_failure': True,
    'email_on_retry': False,
    'retries': 3,
    'retry_delay': timedelta(minutes=5),
}

with DAG(
    'etl_pipeline',
    default_args=default_args,
    description='ETL pipeline for customer data',
    schedule_interval='@daily',
    catchup=False,
) as dag:

    extract_task = PythonOperator(
        task_id='extract_data',
        python_callable=extract_data,
    )

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

    load_task = SnowflakeOperator(
        task_id='load_to_warehouse',
        sql='sql/load_into_warehouse.sql',
        snowflake_conn_id='snowflake_conn',
        warehouse='COMPUTE_WH',
        database='ANALYTICS',
        schema='PUBLIC',
    )

    # Define dependencies
    extract_task >> transform_task >> load_task