# 1. Read the last watermark from a control table
watermark_row = spark.read.format("delta").load("/mnt/control/watermarks") \
    .filter("pipeline_id = 'orders_incremental'") \
    .collect()[0]
last_ts = watermark_row["last_watermark_value"]

# 2. Pull only changed rows since that timestamp
incremental_df = spark.read.format("jdbc") \
    .option("url", jdbc_url) \
    .option("dbtable",
        f"(SELECT * FROM orders WHERE updated_at > '{last_ts}') t") \
    .load()

# 3. Process, then write the new max as the next watermark
new_max = incremental_df.agg({"updated_at": "max"}).collect()[0][0]

from datetime import datetime

low_watermark = "2026-06-14 02:00:00"      # from control table
high_watermark = datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S")  # captured NOW, before read

query = f"""
    SELECT * FROM orders
    WHERE updated_at > '{low_watermark}'
      AND updated_at <= '{high_watermark}'
"""
incremental_df = spark.read.format("jdbc").option("dbtable", f"({query}) t").load()
# After success, high_watermark becomes the next run's low_watermark

from delta.tables import DeltaTable

target = DeltaTable.forPath(spark, "/mnt/gold/dim_customer")
source_df = spark.read.format("delta").load("/mnt/silver/customer_updates")

target.alias("tgt").merge(
        source_df.alias("src"),
        "tgt.customer_id = src.customer_id"
    ) \
    .whenMatchedUpdateAll() \    # overwrite ALL matching columns — no history kept
    .whenNotMatchedInsertAll() \
    .execute()

# dropDuplicates() with no arguments compares ALL columns
deduped_df = raw_df.dropDuplicates()

# Equivalent using distinct()
deduped_df = raw_df.distinct()

from pyspark.sql.functions import col, to_date

# Identify late facts: event_date older than today's processing partition
incoming_df = spark.read.format("delta").load("/mnt/bronze/sales/orders_today")
today = "2026-06-15"

on_time = incoming_df.filter(to_date(col("order_date")) == today)
late_facts = incoming_df.filter(to_date(col("order_date")) != today)

# On-time facts: simple append to today's partition
on_time.write.format("delta").mode("append") \
    .partitionBy("order_date").save("/mnt/silver/orders")

# Late facts: MERGE into their HISTORICAL partition, then flag for downstream re-aggregation
from delta.tables import DeltaTable
target = DeltaTable.forPath(spark, "/mnt/silver/orders")
target.alias("tgt").merge(
        late_facts.alias("src"),
        "tgt.order_id = src.order_id"
    ).whenNotMatchedInsertAll().execute()

# Write affected dates to a "reprocessing queue" table for downstream Gold jobs
affected_dates = late_facts.select(to_date("order_date").alias("affected_date")).distinct()
affected_dates.write.format("delta").mode("append").save("/mnt/control/reprocess_queue")

# Replacing a specific partition is idempotent — running this 1x or 10x
# for the same order_date produces the identical final state
processed_df.write.format("delta") \
    .mode("overwrite") \
    .option("replaceWhere", "order_date = '2026-06-15'") \
    .save("/mnt/silver/orders")

def write_idempotent(batch_df, batch_id):
    # Check the control table first
    already_processed = spark.read.format("delta").load("/mnt/control/processed_batches") \
        .filter(f"batch_id = {batch_id} AND pipeline_name = 'orders_stream'") \
        .count() > 0

    if already_processed:
        return  # skip — this batch was already written, do nothing

    # Write data + log the batch ID in the SAME transaction (Delta MERGE supports this)
    batch_df.write.format("delta").mode("append").save("/mnt/silver/orders")

    spark.createDataFrame([(batch_id, "orders_stream")], ["batch_id", "pipeline_name"]) \
        .write.format("delta").mode("append").save("/mnt/control/processed_batches")

stream_df.writeStream.foreachBatch(write_idempotent) \
    .option("checkpointLocation", "/mnt/checkpoints/orders_stream") \
    .start()

import time
import random
from functools import wraps

def retry_with_backoff(max_retries=5, base_delay=1):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    if attempt == max_retries - 1:
                        raise  # exhausted retries — propagate the error
                    delay = base_delay * (2 ** attempt) + random.uniform(0, 1)  # + jitter
                    print(f"Attempt {attempt+1} failed: {e}. Retrying in {delay:.1f}s...")
                    time.sleep(delay)
        return wrapper
    return decorator

@retry_with_backoff(max_retries=4, base_delay=2)
def fetch_from_source_api():
    response = requests.get("https://api.source-system.com/orders", timeout=30)
    response.raise_for_status()
    return response.json()

from pyspark.sql.functions import col

# Read with permissive mode — captures unparseable rows in _corrupt_record
raw_df = spark.read.option("mode", "PERMISSIVE") \
    .option("columnNameOfCorruptRecord", "_corrupt_record") \
    .json("/mnt/raw/orders/*.json")

good_records = raw_df.filter(col("_corrupt_record").isNull()).drop("_corrupt_record")
bad_records = raw_df.filter(col("_corrupt_record").isNotNull())

# Good records continue down the normal pipeline
good_records.write.format("delta").mode("append").save("/mnt/bronze/orders")

# Bad records go to the DLQ with metadata for triage
bad_records.withColumn("rejected_at", current_timestamp()) \
    .withColumn("source_file", lit("orders/*.json")) \
    .write.format("delta").mode("append").save("/mnt/quarantine/orders_dlq")

from pyspark.sql.functions import current_timestamp, lit

batch_id = "20260615_020000"
run_id = "run_7f3a9c"

audited_df = transformed_df \
    .withColumn("created_timestamp", current_timestamp()) \
    .withColumn("updated_timestamp", current_timestamp()) \
    .withColumn("batch_id", lit(batch_id)) \
    .withColumn("run_id", lit(run_id))

from pyspark.sql.functions import col

null_pk_count = df.filter(col("order_id").isNull()).count()
if null_pk_count > 0:
    raise ValueError(f"Data Quality FAILURE: {null_pk_count} rows have null order_id")

total_count = df.count()
distinct_count = df.select("order_id").distinct().count()

if total_count != distinct_count:
    raise ValueError(f"Duplicate order_ids found: {total_count - distinct_count} duplicates")

# left_anti join finds fact rows with NO matching dimension row
orphans = fact_orders.join(dim_customer, "customer_id", "left_anti")
orphan_count = orphans.count()

if orphan_count > 0:
    print(f"WARNING: {orphan_count} orders reference customers not in dim_customer")
    # could route these to DLQ, or auto-create placeholder dims (see 23.5 Late Dimensions)

from pyspark.sql.functions import col

invalid_dates = df.filter(col("ship_date") < col("order_date"))
invalid_discount = df.filter((col("discount_pct") < 0) | (col("discount_pct") > 100))

if invalid_dates.count() > 0 or invalid_discount.count() > 0:
    raise ValueError("Business rule violations detected — see quarantine table")

today_count = df.count()
avg_7day = 450000  # fetched from historical run metrics in audit table

lower_bound = avg_7day * 0.8
upper_bound = avg_7day * 1.2

if not (lower_bound <= today_count <= upper_bound):
    print(f"ALERT: today's count {today_count} is outside expected range "
          f"[{lower_bound:.0f}, {upper_bound:.0f}]")

import sys
from pyspark.sql.functions import col

# Accept a date range as parameters — same code works for daily runs AND backfills
process_date = sys.argv[1]  # e.g. "2026-06-15", passed by Airflow/orchestrator

source_df = spark.read.format("delta").load("/mnt/bronze/orders") \
    .filter(col("order_date") == process_date)

result_df = transform(source_df)

# Idempotent write — replaying the same date overwrites cleanly
result_df.write.format("delta") \
    .mode("overwrite") \
    .option("replaceWhere", f"order_date = '{process_date}'") \
    .save("/mnt/silver/orders")

# A backfill is just: for date in date_range: spark-submit job.py {date}

source_count = source_df.count()
target_count = spark.read.format("delta").load("/mnt/silver/orders") \
    .filter(f"order_date = '{process_date}'").count()

diff = target_count - source_count
status = "PASS" if diff == 0 else "FAIL"

recon_row = (process_date, "orders_load", source_count, target_count, diff, status)
spark.createDataFrame([recon_row],
    ["run_date", "pipeline_id", "source_count", "target_count", "diff", "status"]) \
    .write.format("delta").mode("append").save("/mnt/control/reconciliation_report")

from pyspark.sql.functions import sum as _sum

source_total = source_df.agg(_sum("order_amount")).collect()[0][0]
target_total = target_df.agg(_sum("order_amount")).collect()[0][0]

# Use a small tolerance for floating-point comparisons, exact match for Decimal
if abs(source_total - target_total) > 0.01:
    print(f"CONTROL TOTAL MISMATCH: source={source_total}, target={target_total}")

from pyspark.sql.functions import sha2, col, regexp_replace, lit

sanitized_df = raw_df \
    .withColumn("email_hash", sha2(col("email"), 256)) \
    .withColumn("email", lit("***MASKED***")) \
    .withColumn("phone", regexp_replace(col("phone"), r"^\d{6}", "XXXXXX")) \
    .withColumn("ssn", lit("XXX-XX-XXXX")) \
    .withColumn("age_band",
        (col("age") / 10).cast("int") * 10) \
    .drop("age")

from delta.tables import DeltaTable

# Step 1: Delete the row (or overwrite PII columns with nulls for soft-erasure)
target = DeltaTable.forPath(spark, "/mnt/gold/dim_customer")
target.delete("customer_id = '501'")

# Step 2: VACUUM removes old Parquet files after retention window
# Without VACUUM, the deleted row is still recoverable via time travel
target.vacuum(0)  # 0 hours = immediate (use cautiously; disables time-travel for this table)

# Better practice: set a short retention on PII tables, log the erasure request
spark.sql("ALTER TABLE gold.dim_customer SET TBLPROPERTIES ('delta.logRetentionDuration'='interval 7 days')")

from pyspark.sql.functions import col, current_timestamp, lit, to_json, struct

# Identify null-PK records
bad_pk = incoming_df.filter(col("order_id").isNull())
good_records = incoming_df.filter(col("order_id").isNotNull())

# Enrich with quarantine metadata and route to quarantine table
quarantine_df = bad_pk \
    .withColumn("raw_record", to_json(struct("*"))) \
    .withColumn("error_code", lit("NULL_PK")) \
    .withColumn("error_message", lit("order_id is null — cannot process")) \
    .withColumn("source_pipeline", lit("orders_silver_load")) \
    .withColumn("batch_id", lit(batch_id)) \
    .withColumn("quarantine_timestamp", current_timestamp()) \
    .withColumn("reprocess_status", lit("PENDING")) \
    .select("raw_record", "error_code", "error_message",
            "source_pipeline", "batch_id", "quarantine_timestamp", "reprocess_status")

quarantine_df.write.format("delta").mode("append").save("/mnt/quarantine/orders")

# Good records continue normally
good_records.write.format("delta").mode("append").save("/mnt/silver/orders")

from pyspark.sql.functions import from_json

quarantine_table = DeltaTable.forPath(spark, "/mnt/quarantine/orders")

# Read PENDING records, re-parse raw_record, attempt re-injection
pending = spark.read.format("delta").load("/mnt/quarantine/orders") \
    .filter("reprocess_status = 'PENDING' AND error_code = 'NULL_PK'")

# Parse the raw JSON back into a DataFrame with the original schema
reparsed = pending.select(from_json(col("raw_record"), original_schema).alias("d")).select("d.*")

# Run through the same validation — if it passes now, write to Silver
fixed = reparsed.filter(col("order_id").isNotNull())
fixed.write.format("delta").mode("append").save("/mnt/silver/orders")

# Update quarantine status
quarantine_table.update(
    condition = "reprocess_status = 'PENDING' AND error_code = 'NULL_PK'",
    set = {"reprocess_status": "'REPROCESSED'"}
)

from pyspark.sql.types import *
from pyspark.sql.functions import current_timestamp, lit
import uuid, time

run_history_schema = StructType([
    StructField("run_id",          StringType()),    # UUID for this specific run
    StructField("pipeline_name",   StringType()),    # e.g. "orders_silver_load"
    StructField("pipeline_version",StringType()),    # git commit hash or semver
    StructField("start_time",       TimestampType()),
    StructField("end_time",         TimestampType()),
    StructField("duration_seconds", LongType()),
    StructField("status",           StringType()),    # RUNNING / SUCCESS / FAILED / PARTIAL
    StructField("source_count",     LongType()),
    StructField("target_count",     LongType()),
    StructField("rejected_count",   LongType()),
    StructField("watermark_low",    StringType()),
    StructField("watermark_high",   StringType()),
    StructField("error_message",    StringType()),    # null if SUCCESS
    StructField("triggered_by",     StringType()),    # airflow / manual / api
])

# At pipeline start — write RUNNING status
run_id = str(uuid.uuid4())
start_ts = time.time()
spark.createDataFrame([(run_id, "orders_silver_load", "v2.3.1",
                        current_timestamp(), None, None, "RUNNING",
                        None, None, None, low_wm, high_wm, None, "airflow")],
                      run_history_schema) \
    .write.format("delta").mode("append").save("/mnt/control/pipeline_run_history")

# ... pipeline runs ...

# At pipeline end — UPDATE the row with final status and metrics
run_history_table = DeltaTable.forPath(spark, "/mnt/control/pipeline_run_history")
run_history_table.update(
    condition = f"run_id = '{run_id}'",
    set = {
        "end_time": "current_timestamp()",
        "duration_seconds": str(int(time.time() - start_ts)),
        "status": "'SUCCESS'",
        "source_count": str(source_count),
        "target_count": str(target_count),
        "rejected_count": str(rejected_count),
    }
)

from datetime import datetime, timedelta

def check_sla(pipeline_name, expected_by_utc, run_date):
    runs = spark.read.format("delta").load("/mnt/control/pipeline_run_history") \
        .filter(f"pipeline_name = '{pipeline_name}'") \
        .filter(f"status = 'SUCCESS'") \
        .filter(f"date(start_time) = '{run_date}'")

    if runs.count() == 0:
        alert(f"SLA BREACH: {pipeline_name} has not completed for {run_date}")
        return

    completion_time = runs.agg({"end_time": "max"}).collect()[0][0]
    if completion_time > expected_by_utc:
        delay_mins = (completion_time - expected_by_utc).seconds // 60
        alert(f"SLA BREACH: {pipeline_name} completed {delay_mins} mins late")

check_sla("orders_gold_daily",
          datetime(2026, 6, 15, 7, 15),  # 07:15 UTC (SLA 07:00 + 15 min grace)
          "2026-06-15")

# After every successful Gold layer write, record the data freshness
freshness_row = spark.createDataFrame([
    ("gold.fact_orders", max_source_event_time, current_timestamp_val, "SUCCESS")
], ["table_name", "max_event_time", "metadata_written_at", "status"])

freshness_row.write.format("delta").mode("append").save("/mnt/control/table_freshness")

# Analysts can query this to understand data currency before running reports:
# SELECT max_event_time FROM control.table_freshness
# WHERE table_name = 'gold.fact_orders'
# ORDER BY metadata_written_at DESC LIMIT 1

from pyspark.sql.functions import avg, stddev, col

history = spark.read.format("delta").load("/mnt/control/pipeline_run_history") \
    .filter("pipeline_name = 'orders_gold_daily' AND status = 'SUCCESS'") \
    .filter("start_time >= current_date() - interval 30 days")

stats = history.agg(
    avg("duration_seconds").alias("avg_duration"),
    stddev("duration_seconds").alias("std_duration")
).collect()[0]

threshold = stats["avg_duration"] + 3 * stats["std_duration"]  # 3-sigma upper bound
if todays_duration > threshold:
    alert(f"PERF ANOMALY: today's run took {todays_duration}s, threshold is {threshold:.0f}s")

# control.pipeline_config table contains one row per managed table:
# pipeline_id | source_type | source_path | target_path | key_columns
# | watermark_col | scd_type | dq_rules_json | is_active

configs = spark.read.format("delta").load("/mnt/control/pipeline_config") \
    .filter("is_active = true").collect()

for cfg in configs:
    try:
        run_pipeline(
            pipeline_id   = cfg["pipeline_id"],
            source_type   = cfg["source_type"],
            source_path   = cfg["source_path"],
            target_path   = cfg["target_path"],
            key_cols      = cfg["key_columns"].split(","),
            watermark_col = cfg["watermark_col"],
            scd_type      = cfg["scd_type"],
            dq_rules      = json.loads(cfg["dq_rules_json"])
        )
    except Exception as e:
        log_failure(cfg["pipeline_id"], str(e))
        # continue to next pipeline — fail-soft pattern

# Adding a NEW table to the platform = INSERT one row into pipeline_config
# No new code. No deployment. Just config.

from delta.tables import DeltaTable
from pyspark.sql import DataFrame

def generic_upsert(source_df: DataFrame, target_path: str, key_cols: list):
    """Merge source_df into target Delta table, matching on key_cols."""

    merge_condition = " AND ".join([f"tgt.{k} = src.{k}" for k in key_cols])

    if DeltaTable.isDeltaTable(spark, target_path):
        target = DeltaTable.forPath(spark, target_path)
        target.alias("tgt").merge(
            source_df.alias("src"),
            merge_condition
        ).whenMatchedUpdateAll().whenNotMatchedInsertAll().execute()
    else:
        # First run — no target table yet, just create it
        source_df.write.format("delta").mode("overwrite").save(target_path)

# Usage — works for ANY table with ANY key columns
generic_upsert(orders_df,    "/mnt/silver/orders",    ["order_id"])
generic_upsert(customers_df, "/mnt/silver/customers", ["customer_id"])
generic_upsert(products_df,  "/mnt/silver/products",  ["product_id", "region_code"])

from pyspark.sql import Window
from pyspark.sql.functions import row_number, col

w = Window.partitionBy("order_id").orderBy(col("updated_at").desc())
deduped = df.withColumn("rn", row_number().over(w)).filter("rn = 1").drop("rn")

df.write.format("delta").mode("overwrite") \
    .option("replaceWhere", "order_date = '2026-06-15'") \
    .save("/mnt/silver/orders")

from delta.tables import DeltaTable
DeltaTable.forPath(spark, target_path).alias("tgt").merge(
    source_df.alias("src"), "tgt.id = src.id"
).whenMatchedUpdateAll().whenNotMatchedInsertAll().execute()

DeltaTable.forPath(spark, target_path).alias("tgt").merge(
    deletes_df.alias("src"), "tgt.customer_id = src.customer_id"
).whenMatchedUpdate(set={"is_deleted": "true", "updated_timestamp": "current_timestamp()"}).execute()