Analytics Architecture and Reporting Questions

MERGE target t
USING (SELECT * EXCEPT(row_num) FROM (
  SELECT *, ROW_NUMBER() OVER (PARTITION BY event_id ORDER BY processed_at DESC) AS row_num
  FROM staging.events_v2
)) s
ON t.event_id = s.event_id
WHEN MATCHED AND s.row_num = 1 THEN UPDATE SET ...
WHEN NOT MATCHED AND s.row_num = 1 THEN INSERT (...)

from airflow import DAG
from airflow.utils.dates import days_ago
from airflow.operators.python import PythonOperator
from airflow.providers.amazon.aws.operators.s3 import S3ListOperator, S3CopyObjectOperator
from airflow.providers.apache.spark.operators.spark_submit import SparkSubmitOperator
from airflow.providers.snowflake.operators.snowflake import SnowflakeOperator
from airflow.providers.postgres.operators.postgres import PostgresOperator
from airflow.utils.email import send_email
from airflow.exceptions import AirflowFailException
import logging

default_args = {
    "owner": "data-engineer",
    "depends_on_past": False,
    "retries": 3,
    "retry_delay": 300,  # 5 minutes
    "email": ["oncall-team@example.com"],
    "email_on_failure": False,  # we handle notification in callback
}

def notify_failure(context):
    dag_id = context.get("dag").dag_id
    task_id = context.get("task_instance").task_id
    exc = context.get("exception")
    subject = f"[ALERT] DAG {dag_id} - task {task_id} failed"
    body = f"Task: {task_id}\nDAG: {dag_id}\nException: {exc}\nLog: {context.get('task_instance').log_url}"
    # send email
    send_email(to=default_args["email"], subject=subject, html_content=body)
    # Optionally post to Slack via webhook (pseudo)
    logging.error("Failure notified: %s", body)

with DAG(
    dag_id="etl_s3_spark_parquet_quality_load",
    default_args=default_args,
    description="S3 -> Spark transform -> DQ -> Load to Redshift/Snowflake",
    schedule_interval="0 2 * * *",  # daily at 02:00
    start_date=days_ago(3),
    catchup=True,              # allow backfills to re-run historical dates
    max_active_runs=1,
    on_failure_callback=notify_failure,
    sla_miss_callback=None,
    tags=["etl", "spark", "data-quality"],
) as dag:

    # 1) Extract: ensure arrival of files in S3 prefix for execution date
    list_s3 = S3ListOperator(
        task_id="list_raw_s3_files",
        bucket="my-data-bucket",
        prefix="{{ ds }}/raw/",
        aws_conn_id="aws_default",
        poke_interval=60,
        timeout=600,
    )

    # Optional: copy to staging area or mark processed
    copy_to_staging = S3CopyObjectOperator(
        task_id="copy_to_staging",
        source_bucket_key="s3://my-data-bucket/{{ ds }}/raw/",
        dest_bucket_key="s3://my-data-bucket/{{ ds }}/staging/",
        aws_conn_id="aws_default",
        replace=True,
    )

    # 2) Transform: Spark job writes partitioned Parquet to data lake (partition by date)
    spark_transform = SparkSubmitOperator(
        task_id="spark_transform_to_parquet",
        application="/opt/airflow/dags/jobs/transform_jobs.py",
        name="transform_job_{{ ds }}",
        conf={"spark.executor.memory": "4g"},
        application_args=["--input", "s3://my-data-bucket/{{ ds }}/staging/", "--output", "s3://my-data-lake/processed/", "--partition", "{{ ds }}"],
        conn_id="spark_default",
        retries=2,
        retry_delay=300,
        sla=3600,  # SLA: task should finish within 1 hour
    )

    # 3) Data quality checks: run tests (row counts, null checks, partition existence)
    def dq_checks(**context):
        # pseudo-implementation: call Great Expectations or run SQL checks
        # raise AirflowFailException on failure to trigger notifications/retries
        # Example checks:
        partition = context["ds"]
        # check existence of parquet partition on S3 or Hive metastore
        # check row count > 0 and no critical nulls
        ok = True  # replace with real checks
        if not ok:
            raise AirflowFailException("Data quality checks failed for partition " + partition)

    run_dq = PythonOperator(
        task_id="run_data_quality_checks",
        python_callable=dq_checks,
        provide_context=True,
        retries=1,
        retry_delay=120,
        sla=1800,
    )

    # 4) Load: either Redshift COPY or Snowflake COPY INTO
    load_redshift = PostgresOperator(
        task_id="load_into_redshift",
        postgres_conn_id="redshift_default",
        sql="""
        BEGIN;
        -- example: use COPY to load parquet via Amazon Spectrum or manifest
        COPY target_table FROM 's3://my-data-lake/processed/{{ ds }}/'
        CREDENTIALS 'aws_iam_role=arn:aws:iam::123456:role/RedshiftCopyRole'
        FORMAT AS PARQUET;
        COMMIT;
        """,
        retries=2,
        retry_delay=300,
        sla=3600,
    )

    load_snowflake = SnowflakeOperator(
        task_id="load_into_snowflake",
        snowflake_conn_id="snowflake_default",
        sql="""
        COPY INTO my_db.my_schema.target_table
        FROM @my_stage/processed/{{ ds }}/
        FILE_FORMAT = (TYPE = PARQUET);
        """,
        retries=2,
        retry_delay=300,
        sla=3600,
        trigger_rule="none_failed_or_skipped",  # only run if previous successful
    )

    # Dependencies
    list_s3 >> copy_to_staging >> spark_transform >> run_dq
    # Choose one loader depending on target: Redshift or Snowflake
    run_dq >> [load_redshift, load_snowflake]