{ "cells": [ { "cell_type": "markdown", "metadata": { "application/vnd.databricks.v1+cell": { "cellMetadata": { "byteLimit": 2048000, "rowLimit": 10000 }, "inputWidgets": {}, "nuid": "4a5f62b7-5d10-4139-a792-7a8d87f98acc", "showTitle": true, "tableResultSettingsMap": {}, "title": "READ ME FIRST" } }, "source": [ "# READ ME FIRST: Your free Lattice24 compute waste assessment\n", "\n", "This notebook measures how much of your Databricks job spend went to runs that failed or timed out, and shows what Lattice24 would offer to help recover it.\n", "\n", "1. Type your company name in the **Customer Name** box at the top.\n", "2. Click **Run all**.\n", "3. Wait for the green **READY** message at the bottom (usually a few minutes).\n", "4. Download your report from the path it prints (Catalog > your catalog > lattice24 > reports).\n", "\n", "**What it reads:** your workspace's own system tables for job runs and billing (system.lakeflow and system.billing). Nothing is sent anywhere. Nothing leaves your workspace unless you choose to share the report.\n", "\n", "**What it writes:** a schema called `lattice24` in the catalog named in the **Output Catalog** box (default `workspace`), with three small tables and one HTML report. You can delete it afterwards.\n", "\n", "**What it costs:** a small amount of your own Databricks compute to run the queries.\n", "\n", "**Questions:** Jim Jardine, jim@lattice24.com, 825-305-0225" ] }, { "cell_type": "code", "execution_count": null, "metadata": { "application/vnd.databricks.v1+cell": { "cellMetadata": { "byteLimit": 2048000, "rowLimit": 10000 }, "inputWidgets": {}, "nuid": "d008d049-f90d-4dc6-95eb-ad13c3146e9c", "showTitle": true, "tableResultSettingsMap": {}, "title": "Widget Setup" } }, "outputs": [], "source": [ "# Settings: type your company name, then click Run all\n", "dbutils.widgets.text(\"customer_name\", \"\", \"Customer Name\")\n", "dbutils.widgets.text(\"lookback_days\", \"90\", \"Lookback Days\")\n", "dbutils.widgets.text(\"output_catalog\", \"workspace\", \"Output Catalog\")\n", "dbutils.widgets.text(\"output_schema\", \"lattice24\", \"Output Schema\")\n", "dbutils.widgets.text(\"order_amount\", \"0\", \"Order Amount (optional)\")\n", "dbutils.widgets.text(\"min_days_to_annualize\", \"30\", \"Min Days to Annualize\")\n", "dbutils.widgets.text(\"data_note\", \"\", \"Data Note (optional)\")\n", "dbutils.widgets.text(\"rep_name\", \"Jim Jardine\", \"Lattice24 Contact\")\n", "dbutils.widgets.text(\"rep_email\", \"jim@lattice24.com\", \"Lattice24 Email\")\n", "dbutils.widgets.text(\"rep_phone\", \"825-305-0225\", \"Lattice24 Phone\")\n", "\n", "# Check results collector: cells F, J and L add to this\n", "check_results = []" ] }, { "cell_type": "code", "execution_count": null, "metadata": { "application/vnd.databricks.v1+cell": { "cellMetadata": { "byteLimit": 2048000, "rowLimit": 10000 }, "inputWidgets": {}, "nuid": "66a90a6d-26db-4615-a852-dc09b9bc2285", "showTitle": true, "tableResultSettingsMap": {}, "title": "A. Setup" } }, "outputs": [], "source": [ "# Cell A: Setup \u2014 read widgets, check system tables, set date variables\n", "from pyspark.sql import functions as F\n", "from pyspark.sql.types import StructType, StructField, IntegerType, DoubleType, StringType\n", "from datetime import datetime, timezone\n", "from collections import defaultdict\n", "\n", "CUSTOMER_NAME = dbutils.widgets.get(\"customer_name\")\n", "LOOKBACK_DAYS = int(dbutils.widgets.get(\"lookback_days\"))\n", "OUTPUT_CATALOG = dbutils.widgets.get(\"output_catalog\")\n", "OUTPUT_SCHEMA = dbutils.widgets.get(\"output_schema\")\n", "if not CUSTOMER_NAME.strip():\n", " print(\"Type your company name in the Customer Name box at the top, then click Run all again.\")\n", " dbutils.notebook.exit(\"Customer name missing\")\n", "\n", "RUN_DATE = datetime.now(timezone.utc).strftime('%Y-%m-%d %H:%M:%S UTC')\n", "\n", "print(f\"Customer: {CUSTOMER_NAME}\")\n", "print(f\"Assessment baseline lookback: {LOOKBACK_DAYS} days\")\n", "print(f\"Output: {OUTPUT_CATALOG}.{OUTPUT_SCHEMA}\")\n", "print(f\"Run date: {RUN_DATE}\")\n", "\n", "# Check system table availability\n", "_system_tables = [\n", " \"system.lakeflow.job_run_timeline\",\n", " \"system.lakeflow.jobs\",\n", " \"system.billing.usage\",\n", " \"system.billing.list_prices\",\n", "]\n", "_missing = []\n", "for _tbl in _system_tables:\n", " try:\n", " spark.sql(f\"SELECT 1 FROM {_tbl} LIMIT 1\")\n", " except Exception:\n", " _missing.append(_tbl)\n", "if _missing:\n", " for _t in _missing:\n", " print(f\"Ask your workspace admin to turn on system tables: {_t}. Nothing else is needed.\")\n", " dbutils.notebook.exit(\"System tables missing\")\n", "else:\n", " print(\"\\nAll required system tables are available.\")\n", "\n", "# Check that results can be saved in the chosen catalog\n", "try:\n", " spark.sql(f\"CREATE SCHEMA IF NOT EXISTS {OUTPUT_CATALOG}.{OUTPUT_SCHEMA}\")\n", "except Exception:\n", " print(f\"This notebook can't save results in the catalog '{OUTPUT_CATALOG}'. \"\n", " \"Type a catalog you can write to in the Output Catalog box (for example 'main'), then click Run all again.\")\n", " dbutils.notebook.exit(\"Output catalog not writable\")" ] }, { "cell_type": "code", "execution_count": null, "metadata": { "application/vnd.databricks.v1+cell": { "cellMetadata": { "byteLimit": 2048000, "rowLimit": 10000 }, "inputWidgets": {}, "nuid": "39fcba04-af64-46f5-b887-571d94d1a445", "showTitle": true, "tableResultSettingsMap": {}, "title": "B. Job Run Data" } }, "outputs": [], "source": [ "# Cell B: Job run data from system.lakeflow.job_run_timeline\n", "# Collapse clock-hour slices to one row per run, join with jobs SCD2 for names\n", "\n", "runs_df = spark.sql(f\"\"\"\n", "WITH latest_jobs AS (\n", " SELECT workspace_id, job_id, name\n", " FROM system.lakeflow.jobs\n", " QUALIFY ROW_NUMBER() OVER (PARTITION BY workspace_id, job_id ORDER BY change_time DESC) = 1\n", "),\n", "runs AS (\n", " SELECT\n", " t.workspace_id,\n", " t.job_id,\n", " t.run_id,\n", " MIN(t.period_start_time) AS run_start_time,\n", " MAX(t.period_end_time) AS run_end_time,\n", " MAX_BY(t.result_state, t.period_end_time) AS result_state,\n", " MAX_BY(t.termination_code, t.period_end_time) AS termination_code,\n", " COALESCE(\n", " NULLIF(MAX(t.run_duration_seconds), 0),\n", " BIGINT(TIMESTAMPDIFF(SECOND, MIN(t.period_start_time), MAX(t.period_end_time)))\n", " ) AS duration_seconds\n", " FROM system.lakeflow.job_run_timeline t\n", " WHERE t.period_end_time > CURRENT_TIMESTAMP() - INTERVAL {LOOKBACK_DAYS} DAYS\n", " GROUP BY t.workspace_id, t.job_id, t.run_id\n", " HAVING result_state IS NOT NULL\n", ")\n", "SELECT\n", " r.workspace_id,\n", " r.job_id,\n", " COALESCE(j.name, 'One-time run ' || r.run_id) AS job_name,\n", " r.run_id,\n", " r.run_start_time,\n", " r.run_end_time,\n", " r.result_state,\n", " r.termination_code,\n", " r.duration_seconds\n", "FROM runs r\n", "LEFT JOIN latest_jobs j USING (workspace_id, job_id)\n", "\"\"\")\n", "\n", "runs_df.createOrReplaceTempView(\"runs_data\")\n", "run_count = runs_df.count()\n", "print(f\"Job runs found (last {LOOKBACK_DAYS} days): {run_count}\")\n", "if run_count > 0:\n", " print(\"\\nBreakdown by result state:\")\n", " runs_df.groupBy(\"result_state\").count().orderBy(\"count\", ascending=False).show()\n", "else:\n", " print(f\"This workspace has no job runs in the last {LOOKBACK_DAYS} days, so there is nothing to measure yet.\")\n", " dbutils.notebook.exit(\"No job runs found\")" ] }, { "cell_type": "code", "execution_count": null, "metadata": { "application/vnd.databricks.v1+cell": { "cellMetadata": { "byteLimit": 2048000, "rowLimit": 10000 }, "inputWidgets": {}, "nuid": "26f30480-87a4-45da-8ca6-10646db714c1", "showTitle": true, "tableResultSettingsMap": {}, "title": "C. Billing Data" } }, "outputs": [], "source": [ "# Cell C: Billing data \u2014 DBUs and list-price cost per run\n", "# Filter to JOBS billing, join with list_prices, aggregate to one row per run\n", "\n", "billing_df = spark.sql(f\"\"\"\n", "WITH run_cost AS (\n", " SELECT\n", " u.workspace_id,\n", " u.usage_metadata.job_id AS job_id,\n", " u.usage_metadata.job_run_id AS run_id,\n", " SUM(u.usage_quantity) AS total_dbus,\n", " SUM(u.usage_quantity * pricing.`default`) AS total_cost_usd\n", " FROM system.billing.usage u\n", " LEFT JOIN system.billing.list_prices pricing\n", " ON u.sku_name = pricing.sku_name\n", " AND pricing.price_start_time <= u.usage_start_time\n", " AND (pricing.price_end_time >= u.usage_start_time OR pricing.price_end_time IS NULL)\n", " AND pricing.currency_code = 'USD'\n", " WHERE u.billing_origin_product = 'JOBS'\n", " AND u.usage_metadata.job_run_id IS NOT NULL\n", " AND u.usage_date >= CURRENT_DATE() - INTERVAL {LOOKBACK_DAYS + 2} DAYS\n", " GROUP BY u.workspace_id, u.usage_metadata.job_id, u.usage_metadata.job_run_id\n", ")\n", "SELECT\n", " r.*,\n", " COALESCE(c.total_dbus, 0.0) AS total_dbus,\n", " COALESCE(c.total_cost_usd, 0.0) AS total_cost_usd\n", "FROM runs_data r\n", "LEFT JOIN run_cost c USING (workspace_id, job_id, run_id)\n", "\"\"\")\n", "\n", "billing_df.createOrReplaceTempView(\"runs_with_cost\")\n", "\n", "matched = billing_df.filter(\"total_dbus > 0\").count()\n", "total = billing_df.count()\n", "match_pct = matched / total * 100 if total > 0 else 0\n", "print(f\"Runs with billing match: {matched} / {total} ({match_pct:.1f}%)\")\n", "if match_pct < 50 and total > 0:\n", " print(\"NOTE: Low billing match rate \u2014 some runs may have $0 cost reported.\")\n", " print(\" This can happen with serverless compute or billing lag.\")" ] }, { "cell_type": "code", "execution_count": null, "metadata": { "application/vnd.databricks.v1+cell": { "cellMetadata": { "byteLimit": 2048000, "rowLimit": 10000 }, "inputWidgets": {}, "nuid": "604e3440-6287-4e2a-8a2e-bf4570e80fb1", "showTitle": true, "tableResultSettingsMap": {}, "title": "D. Waste Summary" } }, "outputs": [], "source": [ "# Cell D: Waste summary per job \u2014 save to assessment_baseline and assessment_totals\n", "\n", "# Per-job waste aggregation\n", "waste_base = spark.sql(\"\"\"\n", "SELECT\n", " job_name,\n", " COUNT(DISTINCT run_id) AS total_runs,\n", " COUNT(DISTINCT CASE WHEN result_state != 'SUCCEEDED' THEN run_id END) AS wasted_runs,\n", " SUM(total_dbus) AS total_dbus,\n", " SUM(CASE WHEN result_state != 'SUCCEEDED' THEN total_dbus ELSE 0 END) AS wasted_dbus,\n", " SUM(total_cost_usd) AS total_cost_usd,\n", " SUM(CASE WHEN result_state != 'SUCCEEDED' THEN total_cost_usd ELSE 0 END) AS wasted_cost_usd,\n", " MIN(run_start_time) AS first_run_date,\n", " MAX(run_end_time) AS last_run_date\n", "FROM runs_with_cost\n", "GROUP BY job_name\n", "\"\"\")\n", "\n", "# Top termination codes per job (Python collection for reliable ordering)\n", "term_rows = spark.sql(\"\"\"\n", " SELECT job_name, COALESCE(termination_code, result_state) AS term_code, COUNT(*) AS cnt\n", " FROM runs_with_cost\n", " WHERE result_state != 'SUCCEEDED'\n", " GROUP BY job_name, COALESCE(termination_code, result_state)\n", " ORDER BY job_name, cnt DESC\n", "\"\"\").collect()\n", "\n", "job_top_codes = {}\n", "job_terms = defaultdict(list)\n", "for row in term_rows:\n", " job_terms[row['job_name']].append(f\"{row['term_code']} ({row['cnt']})\")\n", "for job, codes in job_terms.items():\n", " job_top_codes[job] = ', '.join(codes[:5])\n", "\n", "# Build final waste summary with top_termination_codes and waste_pct\n", "waste_pd = waste_base.toPandas()\n", "waste_pd['top_termination_codes'] = waste_pd['job_name'].map(lambda j: job_top_codes.get(j, '(none)'))\n", "waste_pd['waste_pct_of_cost'] = waste_pd.apply(\n", " lambda r: round(r['wasted_cost_usd'] / r['total_cost_usd'] * 100, 1) if r['total_cost_usd'] > 0 else 0.0,\n", " axis=1\n", ")\n", "waste_pd['total_dbus'] = waste_pd['total_dbus'].astype(float).round(2)\n", "waste_pd['wasted_dbus'] = waste_pd['wasted_dbus'].astype(float).round(2)\n", "waste_pd['total_cost_usd'] = waste_pd['total_cost_usd'].astype(float).round(2)\n", "waste_pd['wasted_cost_usd'] = waste_pd['wasted_cost_usd'].astype(float).round(2)\n", "waste_pd['waste_pct_of_cost'] = waste_pd['waste_pct_of_cost'].astype(float)\n", "waste_pd['first_run_date'] = waste_pd['first_run_date'].astype(str)\n", "waste_pd['last_run_date'] = waste_pd['last_run_date'].astype(str)\n", "waste_pd = waste_pd[['job_name', 'total_runs', 'wasted_runs', 'total_dbus', 'wasted_dbus',\n", " 'total_cost_usd', 'wasted_cost_usd', 'waste_pct_of_cost',\n", " 'top_termination_codes', 'first_run_date', 'last_run_date']]\n", "\n", "# Save to assessment_baseline\n", "spark.sql(f\"CREATE SCHEMA IF NOT EXISTS {OUTPUT_CATALOG}.{OUTPUT_SCHEMA}\")\n", "baseline_spark = spark.createDataFrame(waste_pd)\n", "baseline_spark.write.mode(\"overwrite\").saveAsTable(f\"{OUTPUT_CATALOG}.{OUTPUT_SCHEMA}.assessment_baseline\")\n", "\n", "# Compute assessment_totals (single row)\n", "rwc = spark.table(\"runs_with_cost\")\n", "total_runs_val = rwc.count()\n", "wasted_runs_val = rwc.filter(\"result_state != 'SUCCEEDED'\").count()\n", "total_dbus_val = rwc.agg(F.sum(\"total_dbus\")).collect()[0][0] or 0.0\n", "wasted_dbus_val = rwc.filter(\"result_state != 'SUCCEEDED'\").agg(F.sum(\"total_dbus\")).collect()[0][0] or 0.0\n", "total_cost_val = rwc.agg(F.sum(\"total_cost_usd\")).collect()[0][0] or 0.0\n", "wasted_cost_val = rwc.filter(\"result_state != 'SUCCEEDED'\").agg(F.sum(\"total_cost_usd\")).collect()[0][0] or 0.0\n", "window_start_val = rwc.agg(F.min(\"run_start_time\")).collect()[0][0]\n", "window_end_val = rwc.agg(F.max(\"run_end_time\")).collect()[0][0]\n", "waste_pct_val = wasted_cost_val / total_cost_val * 100 if total_cost_val > 0 else 0.0\n", "\n", "totals_schema = StructType([\n", " StructField(\"total_runs\", IntegerType(), True),\n", " StructField(\"wasted_runs\", IntegerType(), True),\n", " StructField(\"total_dbus\", DoubleType(), True),\n", " StructField(\"wasted_dbus\", DoubleType(), True),\n", " StructField(\"total_cost_usd\", DoubleType(), True),\n", " StructField(\"wasted_cost_usd\", DoubleType(), True),\n", " StructField(\"waste_pct\", DoubleType(), True),\n", " StructField(\"lookback_days\", IntegerType(), True),\n", " StructField(\"window_start\", StringType(), True),\n", " StructField(\"window_end\", StringType(), True),\n", " StructField(\"run_date\", StringType(), True),\n", "])\n", "\n", "totals_df = spark.createDataFrame([(\n", " int(total_runs_val), int(wasted_runs_val),\n", " float(total_dbus_val), float(wasted_dbus_val),\n", " float(total_cost_val), float(wasted_cost_val),\n", " float(waste_pct_val),\n", " int(LOOKBACK_DAYS),\n", " str(window_start_val), str(window_end_val),\n", " RUN_DATE\n", ")], schema=totals_schema)\n", "totals_df.write.mode(\"overwrite\").saveAsTable(f\"{OUTPUT_CATALOG}.{OUTPUT_SCHEMA}.assessment_totals\")\n", "\n", "print(f\"assessment_baseline: {waste_pd.shape[0]} jobs saved to {OUTPUT_CATALOG}.{OUTPUT_SCHEMA}.assessment_baseline\")\n", "print(f\"assessment_totals: 1 row saved to {OUTPUT_CATALOG}.{OUTPUT_SCHEMA}.assessment_totals\")\n", "print(f\"Date window: {window_start_val} to {window_end_val}\")\n", "print(f\"Lookback: {LOOKBACK_DAYS} days\")\n", "print(f\"Run date: {RUN_DATE}\")" ] }, { "cell_type": "code", "execution_count": null, "metadata": { "application/vnd.databricks.v1+cell": { "cellMetadata": { "byteLimit": 2048000, "rowLimit": 10000 }, "inputWidgets": {}, "nuid": "92b8a932-b100-4fa9-96f6-b1e884ec8823", "showTitle": true, "tableResultSettingsMap": {}, "title": "E. Plain-English Summary" } }, "outputs": [], "source": [ "# Cell E: Plain-English summary\n", "\n", "totals = spark.sql(f\"SELECT * FROM {OUTPUT_CATALOG}.{OUTPUT_SCHEMA}.assessment_totals\").collect()[0]\n", "baseline = spark.sql(f\"SELECT job_name, total_runs, wasted_runs, total_dbus, wasted_dbus, total_cost_usd, wasted_cost_usd, waste_pct_of_cost, top_termination_codes, first_run_date, last_run_date FROM {OUTPUT_CATALOG}.{OUTPUT_SCHEMA}.assessment_baseline ORDER BY wasted_cost_usd DESC\")\n", "\n", "tc = totals['total_cost_usd']\n", "wc = totals['wasted_cost_usd']\n", "wp = totals['waste_pct']\n", "tr = totals['total_runs']\n", "wr = totals['wasted_runs']\n", "lb = totals['lookback_days']\n", "\n", "print(\"=\" * 70)\n", "print(f\" COMPUTE WASTE BASELINE \u2014 {CUSTOMER_NAME}\")\n", "print(\"=\" * 70)\n", "print()\n", "print(f\"Over the last {lb} days this workspace spent ${tc:,.2f} at list\")\n", "print(f\"price on jobs. ${wc:,.2f} of that went to runs that failed, timed out,\")\n", "print(f\"or were cancelled. That is {wp:.1f}%.\")\n", "print()\n", "print(f\"Total job runs: {tr}\")\n", "if tr > 0:\n", " print(f\"Wasted runs (non-succeeded): {wr} ({wr/tr*100:.1f}%)\")\n", "print()\n", "\n", "top3 = spark.sql(f\"\"\"\n", " SELECT job_name, wasted_cost_usd, waste_pct_of_cost, wasted_runs, total_runs\n", " FROM {OUTPUT_CATALOG}.{OUTPUT_SCHEMA}.assessment_baseline\n", " ORDER BY wasted_cost_usd DESC\n", " LIMIT 3\n", "\"\"\").collect()\n", "\n", "if top3:\n", " print(\"The three jobs wasting the most were:\")\n", " for i, row in enumerate(top3):\n", " print(f\" {i+1}. {row['job_name']}: ${row['wasted_cost_usd']:,.2f} wasted \"\n", " f\"({row['waste_pct_of_cost']:.1f}% of job cost, \"\n", " f\"{row['wasted_runs']} of {row['total_runs']} runs)\")\n", "else:\n", " print(\"No jobs found in the assessment window.\")\n", "\n", "print()\n", "print(\"Full assessment_baseline table (sorted by wasted cost):\")\n", "baseline.show(truncate=False)" ] }, { "cell_type": "code", "execution_count": null, "metadata": { "application/vnd.databricks.v1+cell": { "cellMetadata": { "byteLimit": 2048000, "rowLimit": 10000 }, "inputWidgets": {}, "nuid": "77316d1c-ae92-4c1b-9c58-7e49bd532522", "showTitle": true, "tableResultSettingsMap": {}, "title": "F. Sanity Checks" } }, "outputs": [], "source": [ "# Cell F: Sanity checks \u2014 PASS or FAIL\n", "\n", "print(\"=\" * 70)\n", "print(\" SANITY CHECKS\")\n", "print(\"=\" * 70)\n", "print()\n", "\n", "all_pass = True\n", "\n", "# Check 1: Wasted cost <= total cost\n", "t = spark.sql(f\"SELECT * FROM {OUTPUT_CATALOG}.{OUTPUT_SCHEMA}.assessment_totals\").collect()[0]\n", "if t['wasted_cost_usd'] <= t['total_cost_usd']:\n", " print(f\" PASS: Wasted cost (${t['wasted_cost_usd']:,.2f}) <= Total cost (${t['total_cost_usd']:,.2f})\")\n", "else:\n", " print(f\" FAIL: Wasted cost (${t['wasted_cost_usd']:,.2f}) > Total cost (${t['total_cost_usd']:,.2f})\")\n", " all_pass = False\n", "\n", "# Check 2: Job run counts add up\n", "bl = spark.sql(f\"SELECT SUM(total_runs) AS s_tr, SUM(wasted_runs) AS s_wr FROM {OUTPUT_CATALOG}.{OUTPUT_SCHEMA}.assessment_baseline\").collect()[0]\n", "if bl['s_tr'] == t['total_runs'] and bl['s_wr'] == t['wasted_runs']:\n", " print(f\" PASS: Job run counts add up ({bl['s_tr']} total, {bl['s_wr']} wasted)\")\n", "else:\n", " print(f\" FAIL: Job run counts don't add up \u2014 baseline sums to {bl['s_tr']}/{bl['s_wr']}, \"\n", " f\"totals say {t['total_runs']}/{t['wasted_runs']}\")\n", " all_pass = False\n", "\n", "# Check 3: Billing match rate\n", "matched = spark.sql(\"SELECT COUNT(*) AS c FROM runs_with_cost WHERE total_dbus > 0\").collect()[0]['c']\n", "total_r = spark.sql(\"SELECT COUNT(*) AS c FROM runs_with_cost\").collect()[0]['c']\n", "match_rate = matched / total_r * 100 if total_r > 0 else 0\n", "if match_rate >= 50:\n", " print(f\" PASS: Billing match rate {match_rate:.1f}% ({matched}/{total_r} runs have billing data)\")\n", "elif match_rate > 0:\n", " print(f\" PASS (with note): Billing match rate {match_rate:.1f}% ({matched}/{total_r} runs have billing data)\")\n", " print(f\" Some runs have $0 cost \u2014 billing data may lag or use serverless credits.\")\n", "else:\n", " print(f\" FAIL: Billing match rate 0% \u2014 no runs have billing data.\")\n", " all_pass = False\n", "\n", "# Check 4: Date window \u2014 informational NOTE, not a PASS/FAIL check\n", "ws = t['window_start']\n", "we = t['window_end']\n", "try:\n", " for fmt in ('%Y-%m-%d %H:%M:%S.%f', '%Y-%m-%d %H:%M:%S', '%Y-%m-%d'):\n", " try:\n", " start_dt = datetime.strptime(ws.split('+')[0].strip(), fmt)\n", " end_dt = datetime.strptime(we.split('+')[0].strip(), fmt)\n", " break\n", " except ValueError:\n", " continue\n", " else:\n", " start_dt = end_dt = None\n", " if start_dt and end_dt:\n", " actual_days = (end_dt - start_dt).days\n", " print(f\" NOTE: data covers {actual_days} days of the {t['lookback_days']}-day lookback, \"\n", " f\"because the workspace only has runs from those dates.\")\n", " else:\n", " print(f\" NOTE: Date window recorded as {ws} to {we}\")\n", "except Exception:\n", " print(f\" NOTE: Date window recorded as {ws} to {we}\")\n", "\n", "print()\n", "if all_pass:\n", " print(\" ALL CHECKS PASSED\")\n", "else:\n", " print(\" SOME CHECKS FAILED \u2014 see above\")\n", "\n", "check_results.append((\"Sanity checks\", all_pass))" ] }, { "cell_type": "code", "execution_count": null, "metadata": { "application/vnd.databricks.v1+cell": { "cellMetadata": { "byteLimit": 2048000, "rowLimit": 10000 }, "inputWidgets": {}, "nuid": "3d37f891-934e-41ba-b88e-954d7af0681b", "showTitle": true, "tableResultSettingsMap": {}, "title": "G. Fixable Waste" } }, "outputs": [], "source": [ "# Cell G: Fixable waste \u2014 FAILED and TIMED_OUT runs (not CANCELLED)\n", "\n", "ORDER_AMOUNT = float(dbutils.widgets.get(\"order_amount\"))\n", "MIN_DAYS_TO_ANNUALIZE = int(dbutils.widgets.get(\"min_days_to_annualize\"))\n", "\n", "# Show distinct result_state values in the runs data\n", "distinct_states = [r['result_state'] for r in\n", " spark.sql(\"SELECT DISTINCT result_state FROM runs_with_cost ORDER BY result_state\").collect()]\n", "print(\"Distinct result_state values in runs data:\")\n", "for s in distinct_states:\n", " print(f\" {s}\")\n", "\n", "# Map actual result_state values to three groups\n", "# Fixable: FAILED, TIMED_OUT\n", "# Not fixable: CANCELLED (a person stopped the run on purpose)\n", "state_to_group = {\n", " 'SUCCEEDED': None,\n", " 'FAILED': 'FAILED',\n", " 'TIMEDOUT': 'TIMED_OUT',\n", " 'CANCELED': 'CANCELLED',\n", " 'ERROR': 'FAILED',\n", " 'UNEXPECTED_ERROR': 'FAILED',\n", "}\n", "\n", "print(\"\\nMapping (actual \u2192 group):\")\n", "for s in distinct_states:\n", " g = state_to_group.get(s, 'FAILED')\n", " if g is None:\n", " print(f\" {s} \u2192 succeeded (not wasted)\")\n", " else:\n", " print(f\" {s} \u2192 {g}\")\n", "\n", "# Determine which states in the data are fixable\n", "fixable_states = [s for s in distinct_states if state_to_group.get(s, 'FAILED') in ('FAILED', 'TIMED_OUT')]\n", "\n", "if fixable_states:\n", " fixable_filter = \"result_state IN (\" + \", \".join([f\"'{s}'\" for s in fixable_states]) + \")\"\n", "else:\n", " fixable_filter = \"FALSE\"\n", "\n", "# Compute fixable waste per job\n", "fixable_df = spark.sql(f\"\"\"\n", "SELECT\n", " job_name,\n", " COUNT(DISTINCT CASE WHEN {fixable_filter} THEN run_id END) AS fixable_runs,\n", " SUM(CASE WHEN {fixable_filter} THEN total_cost_usd ELSE 0 END) AS fixable_cost_usd\n", "FROM runs_with_cost\n", "GROUP BY job_name\n", "\"\"\")\n", "\n", "# Join with existing assessment_baseline and add fixable columns\n", "baseline = spark.table(f\"{OUTPUT_CATALOG}.{OUTPUT_SCHEMA}.assessment_baseline\")\n", "# Drop existing fixable columns if present (idempotent on re-run)\n", "if \"fixable_runs\" in baseline.columns:\n", " baseline = baseline.drop(\"fixable_runs\", \"fixable_cost_usd\")\n", "baseline = baseline.join(fixable_df, \"job_name\", \"left\").select(\n", " \"job_name\", \"total_runs\", \"wasted_runs\", \"total_dbus\", \"wasted_dbus\",\n", " \"total_cost_usd\", \"wasted_cost_usd\", \"waste_pct_of_cost\",\n", " \"top_termination_codes\", \"first_run_date\", \"last_run_date\",\n", " F.coalesce(F.col(\"fixable_runs\"), F.lit(0)).alias(\"fixable_runs\"),\n", " F.coalesce(F.col(\"fixable_cost_usd\").cast(\"double\"), F.lit(0.0)).alias(\"fixable_cost_usd\")\n", ")\n", "\n", "baseline.write.mode(\"overwrite\").option(\"overwriteSchema\", \"true\").saveAsTable(f\"{OUTPUT_CATALOG}.{OUTPUT_SCHEMA}.assessment_baseline\")\n", "\n", "print(f\"\\nUpdated {OUTPUT_CATALOG}.{OUTPUT_SCHEMA}.assessment_baseline with fixable_runs and fixable_cost_usd\")\n", "print(\"\\nFixable waste by job:\")\n", "baseline.select(\"job_name\", \"total_runs\", \"wasted_runs\", \"fixable_runs\",\n", " \"wasted_cost_usd\", F.round(\"fixable_cost_usd\", 2).alias(\"fixable_cost_usd\")).orderBy(\"fixable_cost_usd\", ascending=False).show(truncate=False)" ] }, { "cell_type": "code", "execution_count": null, "metadata": { "application/vnd.databricks.v1+cell": { "cellMetadata": { "byteLimit": 2048000, "rowLimit": 10000 }, "inputWidgets": {}, "nuid": "a72ade25-7e5f-4abc-8dc1-05db3bbaf2d2", "showTitle": true, "tableResultSettingsMap": {}, "title": "H. The Offer" } }, "outputs": [], "source": [ "# Cell H: The offer \u2014 calculated from measured numbers only\n", "\n", "# 1. measured_fixable = total fixable list-price cost in the data window\n", "_row = spark.sql(f\"SELECT SUM(fixable_cost_usd) AS total FROM {OUTPUT_CATALOG}.{OUTPUT_SCHEMA}.assessment_baseline\").collect()[0]\n", "measured_fixable = float(_row['total']) if _row['total'] is not None else 0.0\n", "\n", "# 2. window_days = actual span of data in days with decimals\n", "_t = spark.sql(f\"SELECT * FROM {OUTPUT_CATALOG}.{OUTPUT_SCHEMA}.assessment_totals\").collect()[0]\n", "ws = _t['window_start']\n", "we = _t['window_end']\n", "\n", "for fmt in ('%Y-%m-%d %H:%M:%S.%f', '%Y-%m-%d %H:%M:%S', '%Y-%m-%d'):\n", " try:\n", " start_dt = datetime.strptime(ws.split('+')[0].strip(), fmt)\n", " end_dt = datetime.strptime(we.split('+')[0].strip(), fmt)\n", " break\n", " except ValueError:\n", " continue\n", "\n", "window_days = (end_dt - start_dt).total_seconds() / 86400.0\n", "\n", "# 3. Annualize if window_days >= min_days_to_annualize\n", "if window_days >= MIN_DAYS_TO_ANNUALIZE:\n", " annual_fixable = measured_fixable / window_days * 365\n", " annualized = True\n", " basis_label = \"per year\"\n", "else:\n", " annual_fixable = None\n", " annualized = False\n", " basis_label = \"in the measured window\"\n", " print(f\"Only {window_days:.1f} days of history. An annual figure needs at least {MIN_DAYS_TO_ANNUALIZE} days, so the offer below uses the measured window only.\")\n", "\n", "# 4. basis = annual_fixable if exists, else measured_fixable\n", "basis = annual_fixable if annualized else measured_fixable\n", "\n", "# 5. quoted_savings = 80% of basis\n", "quoted_savings = round(basis * 0.80, 2)\n", "\n", "# 6. customer_share = 75% of quoted_savings, lattice24_share = 25%\n", "lattice24_share = round(quoted_savings * 0.25, 2)\n", "customer_share = round(quoted_savings - lattice24_share, 2)\n", "\n", "# 7. Order coverage\n", "months_to_cover_order = None\n", "if ORDER_AMOUNT > 0:\n", " if annualized:\n", " months_to_cover_order = ORDER_AMOUNT / (lattice24_share / 12)\n", " else:\n", " print(f\"Order coverage needs at least {MIN_DAYS_TO_ANNUALIZE} days of history\")\n", "\n", "# Save to assessment_offer table\n", "offer_schema = StructType([\n", " StructField(\"customer_name\", StringType(), True),\n", " StructField(\"measured_fixable\", DoubleType(), True),\n", " StructField(\"window_days\", DoubleType(), True),\n", " StructField(\"annual_fixable\", DoubleType(), True),\n", " StructField(\"basis\", DoubleType(), True),\n", " StructField(\"basis_label\", StringType(), True),\n", " StructField(\"quoted_savings\", DoubleType(), True),\n", " StructField(\"customer_share\", DoubleType(), True),\n", " StructField(\"lattice24_share\", DoubleType(), True),\n", " StructField(\"order_amount\", DoubleType(), True),\n", " StructField(\"months_to_cover_order\", DoubleType(), True),\n", " StructField(\"window_start\", StringType(), True),\n", " StructField(\"window_end\", StringType(), True),\n", " StructField(\"run_date\", StringType(), True),\n", "])\n", "\n", "offer_df = spark.createDataFrame([(\n", " CUSTOMER_NAME,\n", " measured_fixable,\n", " window_days,\n", " float(annual_fixable) if annual_fixable is not None else None,\n", " float(basis),\n", " basis_label,\n", " float(quoted_savings),\n", " float(customer_share),\n", " float(lattice24_share),\n", " float(ORDER_AMOUNT),\n", " float(months_to_cover_order) if months_to_cover_order is not None else None,\n", " ws, we, RUN_DATE\n", ")], schema=offer_schema)\n", "\n", "offer_df.write.mode(\"overwrite\").saveAsTable(f\"{OUTPUT_CATALOG}.{OUTPUT_SCHEMA}.assessment_offer\")\n", "\n", "print(f\"\\nSaved to {OUTPUT_CATALOG}.{OUTPUT_SCHEMA}.assessment_offer\")" ] }, { "cell_type": "code", "execution_count": null, "metadata": { "application/vnd.databricks.v1+cell": { "cellMetadata": { "byteLimit": 2048000, "rowLimit": 10000 }, "inputWidgets": {}, "nuid": "541786cf-ea5f-4162-a6d7-b34948acdf0e", "showTitle": true, "tableResultSettingsMap": {}, "title": "I. Offer in Plain English" } }, "outputs": [], "source": [ "# Cell I: Print the offer in plain English for the customer\n", "\n", "offer = spark.sql(f\"SELECT * FROM {OUTPUT_CATALOG}.{OUTPUT_SCHEMA}.assessment_offer\").collect()[0]\n", "\n", "basis_label = offer['basis_label']\n", "basis = float(offer['basis'])\n", "quoted = float(offer['quoted_savings'])\n", "customer = float(offer['customer_share'])\n", "lattice24 = float(offer['lattice24_share'])\n", "order_amt = float(offer['order_amount'])\n", "months = offer['months_to_cover_order']\n", "\n", "print(f\"Lattice24 found ${basis:,.2f} in compute that failed or timed out {basis_label}.\")\n", "print(f\"We quote 80% of that: ${quoted:,.2f}.\")\n", "print(f\"You keep 75%: ${customer:,.2f}.\")\n", "print(f\"Lattice24 receives 25%: ${lattice24:,.2f}, and is only paid from savings measured on your own equipment.\")\n", "if order_amt > 0 and months is not None:\n", " print(f\"Lattice24's share pays for your ${order_amt:,.2f} order in about {months:.0f} months.\")" ] }, { "cell_type": "code", "execution_count": null, "metadata": { "application/vnd.databricks.v1+cell": { "cellMetadata": { "byteLimit": 2048000, "rowLimit": 10000 }, "inputWidgets": {}, "nuid": "e100e8c7-1698-4d12-93dd-7a8bde45a488", "showTitle": true, "tableResultSettingsMap": {}, "title": "J. Offer Checks" } }, "outputs": [], "source": [ "# Cell J: Checks \u2014 PASS or FAIL, plus NOTE lines\n", "\n", "offer = spark.sql(f\"SELECT * FROM {OUTPUT_CATALOG}.{OUTPUT_SCHEMA}.assessment_offer\").collect()[0]\n", "_t = spark.sql(f\"SELECT * FROM {OUTPUT_CATALOG}.{OUTPUT_SCHEMA}.assessment_totals\").collect()[0]\n", "\n", "print(\"=\" * 70)\n", "print(\" OFFER CHECKS\")\n", "print(\"=\" * 70)\n", "print()\n", "\n", "all_pass = True\n", "\n", "# Check 1: customer_share + lattice24_share = quoted_savings (to the cent)\n", "cs = float(offer['customer_share'])\n", "ls = float(offer['lattice24_share'])\n", "qs = float(offer['quoted_savings'])\n", "if round(cs + ls, 2) == round(qs, 2):\n", " print(\" PASS: customer_share + lattice24_share = quoted_savings\")\n", "else:\n", " print(f\" FAIL: customer_share + lattice24_share ({cs + ls:.2f}) != quoted_savings ({qs:.2f})\")\n", " all_pass = False\n", "\n", "# Check 2: quoted_savings = 80% of basis\n", "basis = float(offer['basis'])\n", "if round(qs, 2) == round(basis * 0.80, 2):\n", " print(\" PASS: quoted_savings = 80% of basis\")\n", "else:\n", " print(f\" FAIL: quoted_savings ({qs:.2f}) != 80% of basis ({basis * 0.80:.2f})\")\n", " all_pass = False\n", "\n", "# Check 3: fixable cost <= wasted cost from Stage 1\n", "measured_fixable = float(offer['measured_fixable'])\n", "wasted_cost = float(_t['wasted_cost_usd'])\n", "if measured_fixable <= wasted_cost:\n", " print(\" PASS: fixable cost is not more than wasted cost\")\n", "else:\n", " print(f\" FAIL: fixable cost ({measured_fixable:.2f}) > wasted cost ({wasted_cost:.2f})\")\n", " all_pass = False\n", "\n", "# Check 4: measured_fixable matches wasted cost when nothing was cancelled\n", "_cancelled = spark.sql(\"SELECT COUNT(*) AS c FROM runs_with_cost WHERE result_state IN ('CANCELED', 'CANCELLED')\").collect()[0]['c']\n", "if _cancelled == 0:\n", " if abs(measured_fixable - wasted_cost) < 0.01:\n", " print(\" PASS: measured_fixable matches wasted cost (no cancelled runs)\")\n", " else:\n", " print(f\" FAIL: measured_fixable ({measured_fixable:.2f}) != wasted cost ({wasted_cost:.2f})\")\n", " all_pass = False\n", "else:\n", " print(f\" NOTE: {_cancelled} cancelled run(s) excluded from fixable \u2014 measured_fixable not expected to match wasted cost\")\n", "\n", "# NOTE: annual or measured-window\n", "basis_label = offer['basis_label']\n", "window_days = float(offer['window_days'])\n", "if basis_label == \"per year\":\n", " print(f\" NOTE: figures are annual because the data window ({window_days:.1f} days) \"\n", " f\"meets the {MIN_DAYS_TO_ANNUALIZE}-day minimum for annualization.\")\n", "else:\n", " print(f\" NOTE: figures are measured-window only because the data window ({window_days:.1f} days) \"\n", " f\"is less than the {MIN_DAYS_TO_ANNUALIZE}-day minimum for annualization.\")\n", "\n", "print()\n", "if all_pass:\n", " print(\" ALL CHECKS PASSED\")\n", "else:\n", " print(\" SOME CHECKS FAILED \u2014 see above\")\n", "\n", "check_results.append((\"Offer checks\", all_pass))" ] }, { "cell_type": "code", "execution_count": null, "metadata": { "application/vnd.databricks.v1+cell": { "cellMetadata": { "byteLimit": 2048000, "rowLimit": 10000 }, "inputWidgets": {}, "nuid": "6cc794be-1685-4481-acfd-d511580625fe", "showTitle": true, "tableResultSettingsMap": {}, "title": "K. HTML Report" } }, "outputs": [], "source": [ "# Cell K: One-page HTML report\n", "import hashlib\n", "\n", "# Date/duration formatting helpers\n", "def _parse_dt(s):\n", " s = s.strip()\n", " if s.endswith('UTC'):\n", " s = s[:-3].strip()\n", " for fmt in ('%Y-%m-%d %H:%M:%S.%f', '%Y-%m-%d %H:%M:%S', '%Y-%m-%d'):\n", " try:\n", " return datetime.strptime(s, fmt)\n", " except ValueError:\n", " continue\n", " return None\n", "\n", "def _fmt_date(s):\n", " dt = _parse_dt(s)\n", " if dt is None:\n", " return s\n", " return f\"{dt.strftime('%b')} {dt.day}, {dt.year}, {int(dt.strftime('%I'))}:{dt.strftime('%M')} {dt.strftime('%p').lower()} UTC\"\n", "\n", "def _fmt_window(start_s, end_s):\n", " sdt = _parse_dt(start_s)\n", " edt = _parse_dt(end_s)\n", " if sdt is None or edt is None:\n", " return f\"{start_s} to {end_s}\"\n", " start_str = f\"{sdt.strftime('%b')} {sdt.day}, {sdt.year}, {int(sdt.strftime('%I'))}:{sdt.strftime('%M')} {sdt.strftime('%p').lower()}\"\n", " if sdt.date() == edt.date():\n", " end_str = f\"{int(edt.strftime('%I'))}:{edt.strftime('%M')} {edt.strftime('%p').lower()}\"\n", " else:\n", " end_str = f\"{edt.strftime('%b')} {edt.day}, {edt.year}, {int(edt.strftime('%I'))}:{edt.strftime('%M')} {edt.strftime('%p').lower()}\"\n", " return f\"{start_str} \u2013 {end_str} UTC\"\n", "\n", "def _fmt_duration(seconds):\n", " if seconds < 7200:\n", " mins = round(seconds / 60)\n", " return f\"{mins} minute{'s' if mins != 1 else ''}\"\n", " elif seconds < 172800:\n", " hours = seconds / 3600\n", " return f\"{hours:.1f} hours\"\n", " else:\n", " days = seconds / 86400\n", " return f\"{days:.1f} days\"\n", "\n", "# Read all data from tables\n", "_t = spark.sql(f\"SELECT * FROM {OUTPUT_CATALOG}.{OUTPUT_SCHEMA}.assessment_totals\").collect()[0]\n", "_offer = spark.sql(f\"SELECT * FROM {OUTPUT_CATALOG}.{OUTPUT_SCHEMA}.assessment_offer\").collect()[0]\n", "\n", "# Top 5 jobs by fixable cost\n", "top5 = spark.sql(f\"\"\"\n", " SELECT job_name, total_runs, fixable_runs, fixable_cost_usd, waste_pct_of_cost, top_termination_codes\n", " FROM {OUTPUT_CATALOG}.{OUTPUT_SCHEMA}.assessment_baseline\n", " WHERE fixable_cost_usd > 0\n", " ORDER BY fixable_cost_usd DESC\n", " LIMIT 5\n", "\"\"\").collect()\n", "\n", "# Termination code explanations\n", "_term_explain = {\n", " 'RUN_EXECUTION_ERROR': \"the job's code stopped with an error partway through\",\n", " 'RUN_TIMEOUT': \"the job ran past its time limit and was stopped\",\n", " 'TASK_TIMEOUT': \"a task in the job ran past its time limit\",\n", " 'RUNNER_ERROR': \"the compute runner reported an error\",\n", " 'FAILED': \"the job failed\",\n", " 'TIMEDOUT': \"the job timed out\",\n", "}\n", "\n", "def _explain(codes_str):\n", " if not codes_str or codes_str == '(none)':\n", " return \"no failures recorded\"\n", " first = codes_str.split('(')[0].strip()\n", " if not first:\n", " return \"no failures recorded\"\n", " return _term_explain.get(first, f\"the job failed with: {first}\")\n", "\n", "# Offer sentences (same as Cell I, word for word)\n", "basis = float(_offer['basis'])\n", "quoted = float(_offer['quoted_savings'])\n", "customer = float(_offer['customer_share'])\n", "lattice24 = float(_offer['lattice24_share'])\n", "basis_label = _offer['basis_label']\n", "order_amt = float(_offer['order_amount'])\n", "months = _offer['months_to_cover_order']\n", "\n", "offer_sentences = [\n", " f\"Lattice24 found ${basis:,.2f} in compute that failed or timed out {basis_label}.\",\n", " f\"We quote 80% of that: ${quoted:,.2f}.\",\n", " f\"You keep 75%: ${customer:,.2f}.\",\n", " f\"Lattice24 receives 25%: ${lattice24:,.2f}, and is only paid from savings measured on your own equipment.\",\n", "]\n", "if order_amt > 0 and months is not None:\n", " offer_sentences.append(f\"Lattice24's share pays for your ${order_amt:,.2f} order in about {months:.0f} months.\")\n", "\n", "offer_html = ''.join(f'

{s}

' for s in offer_sentences)\n", "\n", "# Data window and totals\n", "total_cost = float(_t['total_cost_usd'])\n", "wasted_cost = float(_t['wasted_cost_usd'])\n", "waste_pct = float(_t['waste_pct'])\n", "run_date = _fmt_date(_t['run_date'])\n", "window_start = _t['window_start']\n", "window_end = _t['window_end']\n", "window_line = _fmt_window(window_start, window_end)\n", "_sdt = _parse_dt(window_start)\n", "_edt = _parse_dt(window_end)\n", "window_seconds = (_edt - _sdt).total_seconds() if _sdt and _edt else float(_offer['window_days']) * 86400\n", "window_days = float(_offer['window_days'])\n", "duration_str = _fmt_duration(window_seconds)\n", "\n", "# Billing match rate\n", "matched = spark.sql(\"SELECT COUNT(*) AS c FROM runs_with_cost WHERE total_dbus > 0\").collect()[0]['c']\n", "total_r = spark.sql(\"SELECT COUNT(*) AS c FROM runs_with_cost\").collect()[0]['c']\n", "billing_match_rate = matched / total_r * 100 if total_r > 0 else 0\n", "\n", "# Data note (optional widget)\n", "data_note = dbutils.widgets.get(\"data_note\").strip()\n", "data_note_html = f'
  • {data_note}
    • ' if data_note else \"\"\n", "\n", "# Rep contact info\n", "rep_name = dbutils.widgets.get(\"rep_name\")\n", "rep_email = dbutils.widgets.get(\"rep_email\")\n", "rep_phone = dbutils.widgets.get(\"rep_phone\")\n", "\n", "# Annualization note\n", "if basis_label == \"per year\":\n", " annual_note = \"Figures are annualized because the window meets the minimum length.\"\n", "else:\n", " annual_note = \"Figures are for the measured window only and are not annualized.\"\n", "\n", "# Top 5 table rows\n", "_table_rows = \"\"\n", "for row in top5:\n", " jn = row['job_name']\n", " tr = row['total_runs']\n", " fr = row['fixable_runs']\n", " fc = float(row['fixable_cost_usd'])\n", " wp = float(row['waste_pct_of_cost'])\n", " tc = row['top_termination_codes']\n", " explanation = _explain(tc)\n", " _table_rows += (\n", " f'\\n \\n'\n", " f' {jn}\\n'\n", " f' {tr}\\n'\n", " f' {fr}\\n'\n", " f' ${fc:,.2f}\\n'\n", " f' {wp:.1f}%\\n'\n", " f' \\n'\n", " f' \\n'\n", " f' {explanation}\\n'\n", " f' '\n", " )\n", "\n", "# SHA-256 fingerprint\n", "_offer_str = str(sorted(_offer.asDict().items()))\n", "_bl_rows = spark.sql(f\"SELECT * FROM {OUTPUT_CATALOG}.{OUTPUT_SCHEMA}.assessment_baseline ORDER BY job_name\").collect()\n", "_bl_str = str([sorted(r.asDict().items()) for r in _bl_rows])\n", "fingerprint = hashlib.sha256((_offer_str + _bl_str).encode()).hexdigest()\n", "\n", "# Build HTML \u2014 CSS as plain string (braces are literal), body as f-string\n", "_css = \"\"\"\"\"\"\n", "\n", "_body = f\"\"\"\n", "

    Lattice24 Compute Waste Assessment

    \n", "
    \n", " {CUSTOMER_NAME}
    \n", " Run date: {run_date}
    \n", " Data window: {window_line}\n", "
    \n", "\n", "
    \n", " In this window, total job spend at list price was ${total_cost:,.2f}.\n", " Of that, ${wasted_cost:,.2f} went to runs that failed or timed out \u2014 {waste_pct:.1f}% of spend.\n", "
    \n", "\n", "

    Where the waste is

    \n", "\n", " \n", " \n", " \n", " \n", " \n", " \n", " {_table_rows}\n", "
    JobRunsFailedFixable costWaste %
    \n", "\n", "

    The offer

    \n", "
    \n", " {offer_html}\n", "
    \n", "\n", "

    How it works

    \n", "
      \n", "
    1. The assessment is free and runs entirely on your own workspace.
      2. \n", "
    2. We quote 80% of what we calculate, so the measured result comes in above the promise.
      3. \n", "
    3. You keep 75% of the savings. Lattice24 takes 25%.
      4. \n", "
    4. Lattice24 is only paid from savings measured on your own equipment.
      5. \n", "
    \n", "\n", "

    What happens next

    \n", "
      \n", "
    1. Fix the top jobs shown above.
      2. \n", "
    2. We measure again each month against this baseline.
      3. \n", "
    3. Each month's savings are split 75/25.
      4. \n", "
    \n", "\n", "
    \n", " \n", "
    \n", "\n", "
    \n", " Questions: {rep_name}, {rep_email}, {rep_phone}
    \n", " Report fingerprint: {fingerprint}\n", "
    \n", "\"\"\"\n", "\n", "REPORT_HTML = f\"\"\"\n", "\n", "\n", "\n", "{_css}\n", "\n", "\n", "{_body}\n", "\n", "\"\"\"" ] }, { "cell_type": "code", "execution_count": null, "metadata": { "application/vnd.databricks.v1+cell": { "cellMetadata": { "byteLimit": 2048000, "rowLimit": 10000 }, "inputWidgets": {}, "nuid": "fd11bcdf-8fb6-4b42-9fb2-44d0763797eb", "showTitle": true, "tableResultSettingsMap": {}, "title": "L. Save and Verify" } }, "outputs": [], "source": [ "# Cell L: Save and show the report\n", "\n", "# Create volume if it doesn't exist\n", "spark.sql(f\"CREATE VOLUME IF NOT EXISTS {OUTPUT_CATALOG}.{OUTPUT_SCHEMA}.reports\")\n", "\n", "# Build filename: Lattice24_Assessment__.html\n", "_today = datetime.now(timezone.utc).strftime('%Y-%m-%d')\n", "_safe_name = CUSTOMER_NAME.replace(' ', '_')\n", "_filename = f\"Lattice24_Assessment_{_safe_name}_{_today}.html\"\n", "_volume_path = f\"/Volumes/{OUTPUT_CATALOG}/{OUTPUT_SCHEMA}/reports/{_filename}\"\n", "\n", "# Save the file\n", "dbutils.fs.put(_volume_path, REPORT_HTML, overwrite=True)\n", "\n", "# Display the page in the notebook\n", "displayHTML(REPORT_HTML)\n", "\n", "# Print the path\n", "print(f\"Report saved to: {_volume_path}\")\n", "\n", "# Checks\n", "print()\n", "print(\"=\" * 70)\n", "print(\" REPORT CHECKS\")\n", "print(\"=\" * 70)\n", "print()\n", "\n", "all_pass = True\n", "\n", "# Check 1: Every dollar figure in the report matches the tables to the cent\n", "_t = spark.sql(f\"SELECT * FROM {OUTPUT_CATALOG}.{OUTPUT_SCHEMA}.assessment_totals\").collect()[0]\n", "_offer = spark.sql(f\"SELECT * FROM {OUTPUT_CATALOG}.{OUTPUT_SCHEMA}.assessment_offer\").collect()[0]\n", "\n", "_expected = [\n", " f\"${float(_t['total_cost_usd']):,.2f}\",\n", " f\"${float(_t['wasted_cost_usd']):,.2f}\",\n", " f\"${float(_offer['basis']):,.2f}\",\n", " f\"${float(_offer['quoted_savings']):,.2f}\",\n", " f\"${float(_offer['customer_share']):,.2f}\",\n", " f\"${float(_offer['lattice24_share']):,.2f}\",\n", "]\n", "# Also check each job's fixable cost from the top-5 table\n", "_top5 = spark.sql(f\"\"\"\n", " SELECT job_name, fixable_cost_usd FROM {OUTPUT_CATALOG}.{OUTPUT_SCHEMA}.assessment_baseline\n", " ORDER BY fixable_cost_usd DESC LIMIT 5\n", "\"\"\").collect()\n", "for row in _top5:\n", " fc = float(row['fixable_cost_usd'])\n", " if fc > 0:\n", " _expected.append(f\"${fc:,.2f}\")\n", "\n", "_missing_amts = [a for a in _expected if a not in REPORT_HTML]\n", "if not _missing_amts:\n", " print(\" PASS: all dollar figures match the tables to the cent\")\n", "else:\n", " print(f\" FAIL: amounts not found in report: {_missing_amts}\")\n", " all_pass = False\n", "\n", "# Check 2: Offer sentences match Cell I exactly\n", "basis = float(_offer['basis'])\n", "quoted = float(_offer['quoted_savings'])\n", "customer = float(_offer['customer_share'])\n", "lattice24 = float(_offer['lattice24_share'])\n", "basis_label = _offer['basis_label']\n", "order_amt = float(_offer['order_amount'])\n", "months = _offer['months_to_cover_order']\n", "\n", "_sentences = [\n", " f\"Lattice24 found ${basis:,.2f} in compute that failed or timed out {basis_label}.\",\n", " f\"We quote 80% of that: ${quoted:,.2f}.\",\n", " f\"You keep 75%: ${customer:,.2f}.\",\n", " f\"Lattice24 receives 25%: ${lattice24:,.2f}, and is only paid from savings measured on your own equipment.\",\n", "]\n", "if order_amt > 0 and months is not None:\n", " _sentences.append(f\"Lattice24's share pays for your ${order_amt:,.2f} order in about {months:.0f} months.\")\n", "\n", "_missing_sent = [s for s in _sentences if s not in REPORT_HTML]\n", "if not _missing_sent:\n", " print(\" PASS: offer sentences match Cell I exactly\")\n", "else:\n", " print(f\" FAIL: sentences not found in report: {_missing_sent}\")\n", " all_pass = False\n", "\n", "# Check 3: No internal words appear in the page\n", "_internal = [\"reserve\", \"commission\"]\n", "_found = [w for w in _internal if w.lower() in REPORT_HTML.lower()]\n", "if not _found:\n", " print(\" PASS: no internal words found in the report\")\n", "else:\n", " print(f\" FAIL: internal words found: {_found}\")\n", " all_pass = False\n", "\n", "print()\n", "if all_pass:\n", " print(\" ALL CHECKS PASSED\")\n", "else:\n", " print(\" SOME CHECKS FAILED \u2014 see above\")\n", "\n", "check_results.append((\"Report checks\", all_pass))" ] }, { "cell_type": "code", "execution_count": null, "metadata": { "application/vnd.databricks.v1+cell": { "cellMetadata": { "byteLimit": 2048000, "rowLimit": 10000 }, "inputWidgets": {}, "nuid": "3bf809c2-9142-4a5b-8b28-f38a9d667d41", "showTitle": true, "tableResultSettingsMap": {}, "title": "M. Readiness Signal" } }, "outputs": [], "source": [ "# Cell M: Final readiness signal\n", "\n", "all_passed = all(result[1] for result in check_results)\n", "\n", "if all_passed:\n", " displayHTML('
    READY
    ')\n", " print()\n", " _today = datetime.now(timezone.utc).strftime('%Y-%m-%d')\n", " _safe_name = CUSTOMER_NAME.replace(' ', '_')\n", " _path = f\"/Volumes/{OUTPUT_CATALOG}/{OUTPUT_SCHEMA}/reports/Lattice24_Assessment_{_safe_name}_{_today}.html\"\n", " print(f\"Report: {_path}\")\n", " print()\n", " _offer = spark.sql(f\"SELECT * FROM {OUTPUT_CATALOG}.{OUTPUT_SCHEMA}.assessment_offer\").collect()[0]\n", " basis = float(_offer['basis'])\n", " quoted = float(_offer['quoted_savings'])\n", " customer = float(_offer['customer_share'])\n", " basis_label = _offer['basis_label']\n", " print(f\"Summary: ${basis:,.2f} in compute went to failed or timed-out runs {basis_label}. Lattice24 quotes ${quoted:,.2f} in savings, and you keep ${customer:,.2f} of it.\")\n", " print()\n", " print(f\"To accept the offer or ask a question, email the report to {dbutils.widgets.get('rep_email')}.\")\n", "else:\n", " displayHTML('
    STOP
    ')\n", " print()\n", " for name, passed in check_results:\n", " if not passed:\n", " print(f\" {name} FAILED.\")\n", " print()\n", " print(f\" Please don't rely on this report. Email the notebook output to {dbutils.widgets.get('rep_email')} and we'll sort it out.\")" ] } ], "metadata": { "application/vnd.databricks.v1+notebook": { "computePreferences": null, "dashboards": [], "environmentMetadata": { "base_environment": "", "environment_version": "6" }, "inputWidgetPreferences": null, "language": "python", "notebookMetadata": { "pythonIndentUnit": 4 }, "notebookName": "Lattice24 Free Assessment", "widgets": {} }, "language_info": { "name": "python" } }, "nbformat": 4, "nbformat_minor": 0 }