12Monitoring

WITH job_name AS (              -- system.lakeflow.jobs is SCD2: one row per EDIT
  SELECT workspace_id, job_id, name, delete_time
  FROM system.lakeflow.jobs
  QUALIFY ROW_NUMBER() OVER (
    PARTITION BY workspace_id, job_id ORDER BY change_time DESC) = 1
),
runs AS (                       -- job_run_timeline emits SEVERAL rows per run
  SELECT
    workspace_id,
    job_id,
    run_id,
    MIN(period_start_time)                                              AS started_at,
    MAX(period_end_time)                                                AS ended_at,
    MAX_BY(result_state, period_end_time)                               AS result_state,
    timestampdiff(SECOND, MIN(period_start_time), MAX(period_end_time)) AS secs
  FROM system.lakeflow.job_run_timeline
  WHERE period_start_time >= current_timestamp() - INTERVAL 14 DAYS
  GROUP BY workspace_id, job_id, run_id
)
SELECT
  n.name,
  COUNT(r.run_id)                                              AS runs,
  SUM(IF(r.result_state <> 'SUCCEEDED', 1, 0))                 AS not_clean,
  ROUND(100 * AVG(IF(r.result_state <> 'SUCCEEDED', 1, 0)), 1) AS pct_not_clean,
  ROUND(AVG(r.secs))                                           AS avg_seconds,
  MAX(r.ended_at)                                              AS last_finished
FROM job_name n                    -- jobs on the LEFT, deliberately: a job that
LEFT JOIN runs r                   -- stopped running must show as runs = 0,
  ON  r.workspace_id = n.workspace_id   -- not vanish from the result
  AND r.job_id       = n.job_id
  AND r.result_state IS NOT NULL   -- terminal runs only; in-flight rows carry NULL.
                                   -- Kept in the JOIN: as a WHERE clause it turns
                                   -- the LEFT JOIN back into an inner one.
WHERE n.delete_time IS NULL        -- deleted jobs keep their run history; skip them
GROUP BY n.name
ORDER BY pct_not_clean DESC NULLS LAST, avg_seconds DESC;

SELECT
  timestamp,
  details:update_progress.state AS state
FROM event_log(TABLE(dwh_prod.silver.sales_order))
WHERE event_type = 'update_progress'
ORDER BY timestamp DESC
LIMIT 20;

-- Once the event log is published to UC, the FROM clause is the only line that
-- changes: swap the TVF for the catalog.schema.name configured on the pipeline.

WITH runs AS (
  SELECT
    job_id,
    run_id,
    MAX(period_end_time)                                                AS ended_at,
    timestampdiff(SECOND, MIN(period_start_time), MAX(period_end_time)) AS secs
  FROM system.lakeflow.job_run_timeline
  WHERE period_start_time >= current_timestamp() - INTERVAL 28 DAYS
  GROUP BY job_id, run_id
  -- healthy runs only: a fast failure would drag the trend DOWN
  HAVING MAX_BY(result_state, period_end_time) = 'SUCCEEDED'
),
drift AS (
  SELECT
    job_id,
    median(CASE WHEN ended_at >= current_timestamp() - INTERVAL 7 DAYS
                THEN secs END) AS p50_7d,
    median(secs)               AS p50_28d
  FROM runs
  GROUP BY job_id
)
SELECT
  job_id,
  ROUND(p50_7d)                         AS p50_7d,
  ROUND(p50_28d)                        AS p50_28d,
  ROUND(p50_7d / NULLIF(p50_28d, 0), 2) AS ratio
FROM drift
WHERE p50_7d > 1.25 * p50_28d
ORDER BY ratio DESC;

      health:
        rules:
          - metric: RUN_DURATION_SECONDS
            op: GREATER_THAN
            value: 5400        # 90 min. Chain starts 03:00, gold is due 06:00.

resources:
  jobs:
    dwh_daily:
      name: dwh_daily
      schedule:
        quartz_cron_expression: "0 0 3 * * ?"
        timezone_id: Europe/Berlin

      # ONE block for the chain. No task below declares notifications.
      webhook_notifications:
        on_failure:
          - id: ${var.oncall_destination}
        on_duration_warning_threshold_exceeded:
          - id: ${var.oncall_destination}
      health:
        rules:
          - metric: RUN_DURATION_SECONDS
            op: GREATER_THAN
            value: 5400
      notification_settings:
        no_alert_for_skipped_runs: true
        no_alert_for_canceled_runs: true

      tasks:
        - task_key: bronze_ingest
          pipeline_task:
            pipeline_id: ${resources.pipelines.bronze_ingest.id}
        - task_key: silver
          depends_on: [{ task_key: bronze_ingest }]
          pipeline_task:
            pipeline_id: ${resources.pipelines.silver.id}
        - task_key: gold
          depends_on: [{ task_key: silver }]
          spark_python_task:
            python_file: ../src/dwh/jobs/gold_fct_order_line.py
        - task_key: assert_gold
          depends_on: [{ task_key: gold }]
          spark_python_task:
            python_file: ../src/dwh/jobs/run_sql_assertions.py

SELECT
  MAX(_tech_loaded_at)                                             AS last_load,
  timestampdiff(MINUTE, MAX(_tech_loaded_at), current_timestamp()) AS minutes_stale
FROM dwh_prod.gold.fct_order_line;

-- Alert condition: minutes_stale > 185  ->  LATE, not FAILED.
--
-- 185 is arithmetic, not taste: this query runs at 06:05 and the chain starts at
-- 03:00, so anything older than those 3h05 was written by LAST night's run.
-- Comparing two timestamps keeps the session time zone out of it. Re-derive the
-- number whenever the schedule moves.

variables:
  oncall_destination:
    description: Notification destination id for the DWH on-call rota.
  dev_null_destination:
    description: Dead-letter destination. Dev and test page nobody, on purpose.
    default: 11112222-3333-4444-5555-666677778888

targets:
  dev:
    variables:
      oncall_destination: ${var.dev_null_destination}
  test:
    variables:
      oncall_destination: ${var.dev_null_destination}
  prod:
    variables:
      oncall_destination: 0a1b2c3d-4e5f-6789-abcd-ef0123456789

# Both variables are DECLARED above. A ${var.x} that was never declared is not a
# runtime surprise — bundle validate rejects it, which is why validate is the
# first step in CI.

SELECT
  LEFT(regexp_replace(statement_text, '\\s+', ' '), 120) AS shape,
  client_application,
  COUNT(*)                                              AS runs,
  ROUND(SUM(total_duration_ms)     / 3.6e6, 2)          AS wall_hours,
  ROUND(SUM(execution_duration_ms) / 3.6e6, 2)          AS exec_hours,
  ROUND(percentile(total_duration_ms, 0.95) / 1000, 1)  AS p95_seconds,
  ROUND(SUM(read_bytes) / POW(1024, 4), 3)              AS read_tib,
  SUM(IF(from_result_cache, 1, 0))                      AS cache_hits,
  SUM(IF(spilled_local_bytes > 0, 1, 0))                AS spilling_runs
FROM system.query.history
WHERE start_time >= current_timestamp() - INTERVAL 7 DAYS
  AND statement_type = 'SELECT'
  AND execution_status = 'FINISHED'
GROUP BY ALL
ORDER BY wall_hours DESC
LIMIT 20;

WITH priced AS (
  SELECT
    u.usage_metadata.job_id  AS job_id,
    u.custom_tags['domain']  AS domain,
    u.usage_date,
    u.usage_quantity * p.pricing.effective_list.default AS list_cost
  FROM system.billing.usage u
  JOIN system.billing.list_prices p
    ON  u.cloud    = p.cloud
    AND u.sku_name = p.sku_name
    AND u.usage_start_time >= p.price_start_time
    AND (p.price_end_time IS NULL OR u.usage_start_time < p.price_end_time)
  WHERE u.usage_date >= current_date() - INTERVAL 36 DAYS
    AND u.usage_date <  current_date() - INTERVAL 1 DAYS   -- closed days only
    AND u.usage_metadata.job_id IS NOT NULL
),
weekly AS (
  SELECT
    domain,
    job_id,
    SUM(IF(usage_date >= current_date() - INTERVAL 8 DAYS, list_cost, 0))     AS last_week,
    SUM(IF(usage_date <  current_date() - INTERVAL 8 DAYS, list_cost, 0)) / 4 AS prior_avg
  FROM priced
  GROUP BY ALL
)
SELECT
  domain,
  job_id,
  ROUND(last_week, 2)                            AS last_week,
  ROUND(prior_avg, 2)                            AS prior_weekly_avg,
  ROUND(100 * (last_week / NULLIF(prior_avg, 0) - 1), 1) AS pct_change
FROM weekly
WHERE prior_avg > 0 AND last_week > 1.25 * prior_avg
ORDER BY pct_change DESC;