Module 16: Production Monitoring & Operations

%sql
SELECT count(*) AS stale_hours
FROM (SELECT timestampdiff(HOUR, max(order_ts), current_timestamp()) AS h FROM orders)
WHERE h > 24

%sql
-- failed runs in the last 7 days
SELECT job_id, run_id, result_state, termination_code, period_start_time
FROM system.lakeflow.job_run_timeline
WHERE period_start_time >= current_date() - INTERVAL 7 DAYS
  AND result_state IN ('FAILED', 'ERROR', 'TIMED_OUT')
ORDER BY period_start_time DESC

def log_run(job, step, status, rows_in=None, rows_out=None, rows_bad=None, message=""):
    (spark.createDataFrame(
        [(job, step, status, rows_in, rows_out, rows_bad, message)],
        "job STRING, step STRING, status STRING, rows_in LONG, rows_out LONG, rows_bad LONG, message STRING")
     .withColumn("run_ts", F.current_timestamp())
     .select("run_ts", "job", "step", "status", "rows_in", "rows_out", "rows_bad", "message")
     .write.mode("append").saveAsTable("ops_run_log"))

src = spark.table("orders")
n_in = src.count()
good = src.filter("customer_id IS NOT NULL AND quantity > 0")
n_out = good.count()
log_run("orders_daily", "clean", "SUCCESS", n_in, n_out, n_in - n_out)

%sql
SELECT * FROM ops_run_log ORDER BY run_ts DESC LIMIT 10

import logging

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s: %(message)s")
log = logging.getLogger("orders_daily")

class DataQualityError(Exception): ...
class SourceUnavailableError(Exception): ...

def step_clean():
    try:
        df = spark.table("orders")
        bad = df.filter("quantity <= 0").count()
        if bad:
            raise DataQualityError(f"{bad} rows with non-positive quantity")
        log.info("clean ok: %d rows", df.count())
    except DataQualityError as e:
        log.error("data problem: %s", e)
        log_run("orders_daily", "clean", "DATA_ERROR", message=str(e))
        raise

checks = spark.sql("""
  SELECT current_timestamp() AS checked_at, 'orders' AS table_name,
         count(*) AS row_count,
         count(*) FILTER (WHERE customer_id IS NULL) AS null_customer,
         count(*) FILTER (WHERE quantity IS NULL)    AS null_quantity,
         count(DISTINCT order_id)                    AS distinct_orders
  FROM orders
""")
checks.write.mode("append").saveAsTable("ops_dq_history")
display(spark.table("ops_dq_history").orderBy("checked_at"))

%sql
SELECT 'orders' AS table_name,
       max(order_ts) AS last_event,
       timestampdiff(HOUR, max(order_ts), current_timestamp()) AS hours_stale,
       timestampdiff(HOUR, max(order_ts), current_timestamp()) <= 24 AS within_slo
FROM orders

import time

def with_retries(fn, attempts=3, delay=10):
    for i in range(1, attempts + 1):
        try:
            return fn()
        except SourceUnavailableError as e:
            log.warning("attempt %d/%d failed: %s", i, attempts, e)
            if i == attempts:
                raise
            time.sleep(delay * i)         # increasing wait

%sql
CREATE OR REPLACE TABLE orders_backfill_copy DEEP CLONE orders

%sql
-- recent Unity Catalog actions by a user
SELECT event_time, action_name, user_identity.email, request_params
FROM system.access.audit
WHERE event_date >= current_date() - INTERVAL 7 DAYS
  AND service_name = 'unityCatalog'
ORDER BY event_time DESC
LIMIT 50

%sql
DESCRIBE HISTORY orders_delta

Runbook: orders_daily

Purpose      Loads and cleans orders. Feeds the revenue dashboard.
Owner        data-eng (channel #data-eng-alerts)
Schedule     Daily 05:00 UTC. SLA: tables ready by 08:00.
Depends on   Landing files in /Volumes/.../orders. Table products.

Alerts
  FAILED          -> check the failed task, see "Common failures"
  Duration > 20m  -> check input volume, query profile

Common failures
  Schema changed      -> Auto Loader stopped: rerun (new column added) or full refresh
  Quarantine > 1%     -> notify the source owner, do NOT backfill yet
  Permission denied   -> check the service principal grants

Recovery
  1. Repair run (reruns failed task and downstream)
  2. Bad data written -> RESTORE TABLE ... TO VERSION AS OF n, then rerun
  3. Backfill: run the job with run_date range; reconcile counts

Escalation   On-call, then team lead after 30 minutes.

%sql
-- usage per job (sum handles correction records)
SELECT usage_metadata.job_id, usage_date, sum(usage_quantity) AS dbus
FROM system.billing.usage
WHERE usage_metadata.job_id IS NOT NULL
GROUP BY ALL
HAVING dbus != 0
ORDER BY usage_date DESC