{ "cells": [ { "cell_type": "markdown", "metadata": { "application/vnd.databricks.v1+cell": { "cellMetadata": {}, "inputWidgets": {}, "nuid": "0191a9a8-acaf-43f1-b5e6-65f5a53557f1", "showTitle": false, "tableResultSettingsMap": {}, "title": "" } }, "source": [ "# The join that should broadcast\n", "\n", "A revenue report joins 7.5 million orders to a customer dimension. The dimension is small enough\n", "to sit in memory, and the plan shuffles it anyway. The report takes minutes, and the total it\n", "prints is wrong.\n", "\n", "Two problems, one cause. The plan is the only thing that shows you both.\n", "\n", "Full instructions: https://brickster.io/labs/the-join-that-should-broadcast\n" ] }, { "cell_type": "markdown", "metadata": { "application/vnd.databricks.v1+cell": { "cellMetadata": {}, "inputWidgets": {}, "nuid": "cc129df2-0723-414d-a16c-606cd4e5189a", "showTitle": false, "tableResultSettingsMap": {}, "title": "" } }, "source": [ "## 1. Your salt\n", "\n", "Copy the eight characters under **Your salt** on the lab page and paste them below. One of the five\n", "answers depends on it.\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": { "application/vnd.databricks.v1+cell": { "cellMetadata": { "byteLimit": 2048000, "rowLimit": 10000 }, "finishTime": 1790329335972, "inputWidgets": {}, "nuid": "923dbcb5-8bb0-4e20-9287-efadb933387c", "showTitle": false, "startTime": 1790329335542, "submitTime": 1790329332548, "tableResultSettingsMap": {}, "title": "" } }, "outputs": [], "source": [ "SALT = \"00000000\" # <- paste yours here\n" ] }, { "cell_type": "markdown", "metadata": { "application/vnd.databricks.v1+cell": { "cellMetadata": {}, "inputWidgets": {}, "nuid": "0d779bc6-af01-4419-b7bc-856e00185da7", "showTitle": false, "tableResultSettingsMap": {}, "title": "" } }, "source": [ "## 2. The dimension you inherited\n", "\n", "`samples.tpch.orders` is the fact table, 7.5 million rows, read in place. The customer dimension\n", "is the part somebody else built: one row per customer per version, because it keeps history.\n", "Every customer has eight versions and exactly one of them is current.\n", "\n", "An ordinary SCD table. Nothing about it looks wrong, until you join it to 7.5 million rows\n", "without asking which versions you actually need.\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": { "application/vnd.databricks.v1+cell": { "cellMetadata": { "byteLimit": 2048000, "rowLimit": 10000 }, "finishTime": 1790341344103, "inputWidgets": {}, "nuid": "a5f3f1c3-2b3d-49de-8cc1-90c34bbaba3b", "showTitle": false, "startTime": 1790341341317, "submitTime": 1790341341279, "tableResultSettingsMap": {}, "title": "" } }, "outputs": [], "source": [ "CATALOG = \"workspace\" # change this if workspace is not where you can create a schema\n", "SCHEMA = \"brickster_broadcast\"\n", "T = f\"{CATALOG}.{SCHEMA}\"\n", "\n", "spark.sql(f\"create schema if not exists {T}\")\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": { "application/vnd.databricks.v1+cell": { "cellMetadata": { "byteLimit": 2048000, "rowLimit": 10000 }, "finishTime": 1790329353525, "inputWidgets": {}, "nuid": "13323820-e85b-4060-a41d-f6efa84745e7", "showTitle": false, "startTime": 1790329343523, "submitTime": 1790329343478, "tableResultSettingsMap": {}, "title": "" } }, "outputs": [], "source": [ "spark.sql(f\"\"\"\n", " create or replace table {T}.customer_scd as\n", " select c.c_custkey, c.c_mktsegment, c.c_nationkey,\n", " t.version,\n", " t.version = 7 as is_current\n", " from samples.tpch.customer c\n", " cross join (select explode(sequence(0, 7)) as version) t\n", "\"\"\")\n", "\n", "print(\"dimension rows:\", spark.table(f\"{T}.customer_scd\").count())\n" ] }, { "cell_type": "markdown", "metadata": { "application/vnd.databricks.v1+cell": { "cellMetadata": {}, "inputWidgets": {}, "nuid": "9b0356df-aa69-4c7c-81ee-4765758bfb66", "showTitle": false, "tableResultSettingsMap": {}, "title": "" } }, "source": [ "## 3. How to read which join you got\n", "\n", "`explain formatted` prints the plan as text, and the only part this lab needs is the name of the\n", "join operator in it. Serverless may prefix it with `Photon`, which is the same operator.\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": { "application/vnd.databricks.v1+cell": { "cellMetadata": { "byteLimit": 2048000, "rowLimit": 10000 }, "finishTime": 1790341348189, "inputWidgets": {}, "nuid": "9188adbe-8b9d-432d-bba2-f48d2e8847aa", "showTitle": true, "startTime": 1790341348030, "submitTime": 1790341347992, "tableResultSettingsMap": {}, "title": "join_node helper" } }, "outputs": [], "source": [ "import re\n", "\n", "JOINS = r\"(?:Photon)?(BroadcastHashJoin|ShuffledHashJoin|ShuffleHashJoin|SortMergeJoin|BroadcastNestedLoopJoin)\"\n", "\n", "\n", "def join_node(sql):\n", " plan = \"\\n\".join(r[0] for r in spark.sql(f\"explain formatted {sql}\").collect())\n", " found = re.findall(JOINS, plan)\n", " return found[0].replace(\"Shuffled\", \"Shuffle\") if found else \"none\"\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": { "application/vnd.databricks.v1+cell": { "cellMetadata": { "byteLimit": 2048000, "rowLimit": 10000 }, "finishTime": 1790341353292, "inputWidgets": {}, "nuid": "5110bb40-e6a4-4f40-a087-f327d718777b", "showTitle": true, "startTime": 1790341353176, "submitTime": 1790341353139, "tableResultSettingsMap": {}, "title": "Plan visualization helpers" } }, "outputs": [], "source": [ "from IPython.display import HTML, display as ipydisplay\n", "\n", "\n", "def _plan_tree(sql):\n", " plan = \"\\n\".join(r[0] for r in spark.sql(f\"explain formatted {sql}\").collect())\n", " lines = plan.split(\"\\n\")\n", " tree_lines = []\n", " for ln in lines:\n", " if ln.startswith(\"(\") or ln.startswith(\"==\") or ln.strip() == \"\":\n", " if tree_lines:\n", " break\n", " continue\n", " tree_lines.append(ln)\n", " return tree_lines\n", "\n", "\n", "def _node_color(text):\n", " t = text.lower()\n", " if \"broadcasthashjoin\" in t:\n", " return \"background:#c8e6c9;border-color:#4caf50\"\n", " if \"shuffledhashjoin\" in t or \"sortmergejoin\" in t:\n", " return \"background:#ffcdd2;border-color:#f44336\"\n", " if \"join\" in t:\n", " return \"background:#e1bee7;border-color:#9c27b0\"\n", " if \"shuffle\" in t:\n", " return \"background:#fff3e0;border-color:#ff9800\"\n", " if \"scan\" in t:\n", " return \"background:#e3f2fd;border-color:#2196f3\"\n", " if \"agg\" in t or \"group\" in t:\n", " return \"background:#fce4ec;border-color:#e91e63\"\n", " if \"project\" in t:\n", " return \"background:#f3e5f5;border-color:#ab47bc\"\n", " return \"background:#f5f5f5;border-color:#bdbdbd\"\n", "\n", "\n", "def _render_tree(tree_lines):\n", " parts = [\"
\"]\n", " for ln in tree_lines:\n", " stripped = ln.lstrip()\n", " indent = len(ln) - len(stripped)\n", " style = _node_color(stripped)\n", " clean = stripped.replace(\":-\", \"|\").replace(\"+-\", \"L\")\n", " parts.append(\n", " \"
\"\n", " \"\".format(indent * 6, style)\n", " + clean\n", " + \"
\"\n", " )\n", " parts.append(\"
\")\n", " return \"\\n\".join(parts)\n", "\n", "\n", "def show_plan(sql, label):\n", " ipydisplay(HTML(\"

\" + label + \"

\"))\n", " ipydisplay(HTML(_render_tree(_plan_tree(sql))))\n" ] }, { "cell_type": "markdown", "metadata": { "application/vnd.databricks.v1+cell": { "cellMetadata": {}, "inputWidgets": {}, "nuid": "905bd8f5-d46a-443f-bcc4-9fe33c64e07f", "showTitle": false, "tableResultSettingsMap": {}, "title": "" } }, "source": [ "## 4. The report, as it runs today\n", "\n", "Revenue per market segment. One join, one group by, nothing exotic.\n", "\n", "The number comes back fast enough to look fine. It is not.\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": { "application/vnd.databricks.v1+cell": { "cellMetadata": { "byteLimit": 2048000, "rowLimit": 10000 }, "finishTime": 1790341377752, "inputWidgets": {}, "nuid": "92f9ac1c-dcae-4d1d-af51-165e2bea6e30", "showTitle": false, "startTime": 1790341359217, "submitTime": 1790341359180, "tableResultSettingsMap": {}, "title": "" } }, "outputs": [], "source": [ "NAIVE = f\"\"\"\n", " select c.c_mktsegment, bigint(round(sum(o.o_totalprice))) as total\n", " from samples.tpch.orders o\n", " join {T}.customer_scd c on c.c_custkey = o.o_custkey\n", " group by 1\n", "\"\"\"\n", "\n", "display(spark.sql(NAIVE))\n" ] }, { "cell_type": "markdown", "metadata": { "application/vnd.databricks.v1+cell": { "cellMetadata": {}, "inputWidgets": {}, "nuid": "e4a995b1-e2ea-435d-b430-aa48732799fc", "showTitle": false, "tableResultSettingsMap": {}, "title": "" } }, "source": [ "## 5. What the plan says about it\n", "\n", "The dimension holds a handful of columns and the report reads three of them. Ask the plan what it\n", "decided to do with that.\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": { "application/vnd.databricks.v1+cell": { "cellMetadata": { "byteLimit": 2048000, "rowLimit": 10000 }, "finishTime": 1790341382844, "inputWidgets": {}, "nuid": "0a9d6c36-9d47-4bee-b0b5-579d36d91b81", "showTitle": false, "startTime": 1790341381573, "submitTime": 1790341381528, "tableResultSettingsMap": {}, "title": "" } }, "outputs": [], "source": [ "# TODO: which join operator does the plan pick for NAIVE. `join_node` is above.\n", "plan_before = None\n", "\n", "print(f\"BRICKSTER:plan_before={plan_before}\")\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": { "application/vnd.databricks.v1+cell": { "cellMetadata": { "byteLimit": 2048000, "rowLimit": 10000 }, "finishTime": 1790341388754, "inputWidgets": {}, "nuid": "0f48b160-b30f-436a-88a8-f7b46fa10d9b", "showTitle": true, "startTime": 1790341387536, "submitTime": 1790341387477, "tableResultSettingsMap": {}, "title": "Visualize naive plan" } }, "outputs": [], "source": [ "show_plan(NAIVE, \"Plan before (naive): \" + plan_before)\n" ] }, { "cell_type": "markdown", "metadata": { "application/vnd.databricks.v1+cell": { "cellMetadata": {}, "inputWidgets": {}, "nuid": "a242bfe6-21b4-4776-ad88-923a8561de45", "showTitle": false, "tableResultSettingsMap": {}, "title": "" } }, "source": [ "## 6. The number the report got wrong\n", "\n", "The join has a second problem, and it is not about speed. Count the rows the join produces against\n", "the rows the fact table holds. A join that keeps every order once comes out at 1.\n", "\n", "Something else comes out at 8. That 8 is sitting in your revenue.\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": { "application/vnd.databricks.v1+cell": { "cellMetadata": { "byteLimit": 2048000, "rowLimit": 10000 }, "finishTime": 1790341776319, "inputWidgets": {}, "nuid": "2a14e966-9b6b-48a9-8cad-7648fc7faca9", "showTitle": false, "startTime": 1790341773687, "submitTime": 1790341773647, "tableResultSettingsMap": {}, "title": "" } }, "outputs": [], "source": [ "# TODO: rows the join produces, divided by rows in samples.tpch.orders, as a whole number.\n", "joined = None\n", "orders = spark.table(\"samples.tpch.orders\").count()\n", "fanout = None\n", "\n", "print(f\"BRICKSTER:fanout={fanout}\")\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": { "application/vnd.databricks.v1+cell": { "cellMetadata": { "byteLimit": 2048000, "rowLimit": 10000 }, "finishTime": 1790342053739, "inputWidgets": {}, "nuid": "e2eb8203-fb41-44ab-b584-a975e616762d", "showTitle": true, "startTime": 1790342049027, "submitTime": 1790342048981, "tableResultSettingsMap": {}, "title": "Illustrate fanout with one order" } }, "outputs": [], "source": [ "# Pick one order and see what the join does to it\n", "sample_order = spark.sql(f\"\"\"\n", " select o_custkey, o_totalprice\n", " from samples.tpch.orders\n", " order by o_custkey\n", " limit 1\n", "\"\"\").collect()[0]\n", "\n", "custkey = sample_order[\"o_custkey\"]\n", "price = sample_order[\"o_totalprice\"]\n", "\n", "print(f\"One order: o_custkey={custkey} o_totalprice={price}\")\n", "print()\n", "\n", "# Without the fix: this order matches ALL 8 versions of its customer\n", "all_matches = spark.sql(f\"\"\"\n", " select c.c_custkey, c.c_mktsegment, c.version, c.is_current\n", " from {T}.customer_scd c\n", " where c.c_custkey = {custkey}\n", " order by c.version\n", "\"\"\")\n", "print(\"Without fix: this ONE order joins to 8 dimension rows (one per version):\")\n", "display(all_matches)\n", "\n", "# With the fix: only the current version survives\n", "print(\"With fix: the same order joins to 1 row (only is_current = true):\")\n", "display(spark.sql(f\"\"\"\n", " select c.c_custkey, c.c_mktsegment, c.version, c.is_current\n", " from {T}.customer_scd c\n", " where c.c_custkey = {custkey} and c.is_current\n", "\"\"\"))\n", "\n", "print(f\"\\nFanout = {fanout}x -> every order's price is counted {fanout} times instead of once.\")\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": { "application/vnd.databricks.v1+cell": { "cellMetadata": { "byteLimit": 2048000, "rowLimit": 10000 }, "finishTime": 1790342440672, "inputWidgets": {}, "nuid": "383a2b66-727c-4eb8-92de-81575266645a", "showTitle": true, "startTime": 1790342438616, "submitTime": 1790342438571, "tableResultSettingsMap": {}, "title": "Why fanout and broadcast are connected" } }, "outputs": [], "source": [ "# Why fanout and broadcast are the same story\n", "all_rows = spark.table(f\"{T}.customer_scd\").count()\n", "current_rows = spark.sql(f\"select count(*) from {T}.customer_scd where is_current\").collect()[0][0]\n", "threshold_mb = 10 # spark.sql.autoBroadcastJoinThreshold default\n", "\n", "print(\"Dimension size:\")\n", "print(f\" All 8 versions: {all_rows:>12,} rows\")\n", "print(f\" Current only: {current_rows:>12,} rows\")\n", "print(f\" Auto-broadcast threshold: {threshold_mb} MB\\n\")\n", "print(\"The 8 versions inflate the table 8x, past the broadcast threshold AND into your revenue.\")\n", "print(\"Filtering to is_current fixes both. The hint makes sure the optimiser agrees.\")\n" ] }, { "cell_type": "markdown", "metadata": { "application/vnd.databricks.v1+cell": { "cellMetadata": {}, "inputWidgets": {}, "nuid": "3f7105c6-fac3-40fc-817e-8682311364d9", "showTitle": false, "tableResultSettingsMap": {}, "title": "" } }, "source": [ "## 7. Make the plan change\n", "\n", "Three things move a join from a shuffle to a broadcast:\n", "\n", "* **Hint**: `/*+ BROADCAST(c) */` tells the optimiser what to do.\n", "* **Statistics**: `ANALYZE TABLE` gives it the size to decide on its own.\n", "* **Less data**: filtering to the rows you actually need.\n", "\n", "The first two change the plan. The third changes the plan *and* the number.\n", "\n", "Write the query you would ship, and ask the plan what it decided.\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": { "application/vnd.databricks.v1+cell": { "cellMetadata": { "byteLimit": 2048000, "rowLimit": 10000 }, "finishTime": 1790341394557, "inputWidgets": {}, "nuid": "7a03a5f8-ef11-4700-a59f-a8735dc4fd7e", "showTitle": true, "startTime": 1790341392656, "submitTime": 1790341392620, "tableResultSettingsMap": {}, "title": "Fixed query with broadcast hint" } }, "outputs": [], "source": [ "# TODO: write the report you would ship, then read its join operator off the plan.\n", "# A hint looks like /*+ BROADCAST(c) */. Statistics come from `analyze table`. The third way\n", "# does not ask the optimiser for anything, and it is the one that also moves the total.\n", "FIXED = None\n", "plan_after = None\n", "\n", "print(f\"BRICKSTER:plan_after={plan_after}\")\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": { "application/vnd.databricks.v1+cell": { "cellMetadata": { "byteLimit": 2048000, "rowLimit": 10000 }, "finishTime": 1790341400164, "inputWidgets": {}, "nuid": "0a7ab1a3-1f68-4156-8137-26093a0a90fe", "showTitle": true, "startTime": 1790341399146, "submitTime": 1790341399107, "tableResultSettingsMap": {}, "title": "Visualize fixed plan" } }, "outputs": [], "source": [ "show_plan(FIXED, \"Plan after (fixed): \" + plan_after)\n" ] }, { "cell_type": "markdown", "metadata": { "application/vnd.databricks.v1+cell": { "cellMetadata": {}, "inputWidgets": {}, "nuid": "ea9fe4c7-3782-4c56-922e-7ef73eb62360", "showTitle": false, "tableResultSettingsMap": {}, "title": "" } }, "source": [ "## 8. The total, once the join is right\n", "\n", "The whole point of the fix is that this number is the revenue the business actually has.\n", "If your rewrite changed it, the rewrite is not the answer. It is a different bug.\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": { "application/vnd.databricks.v1+cell": { "cellMetadata": { "byteLimit": 2048000, "rowLimit": 10000 }, "finishTime": 1790341406680, "inputWidgets": {}, "nuid": "551898ed-6025-49bb-8e0f-1b37417f7768", "showTitle": false, "startTime": 1790341404163, "submitTime": 1790341404128, "tableResultSettingsMap": {}, "title": "" } }, "outputs": [], "source": [ "# TODO: total revenue across every segment, from your fixed query.\n", "total_price = None\n", "\n", "print(f\"BRICKSTER:total_price={total_price}\")\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": { "application/vnd.databricks.v1+cell": { "cellMetadata": { "byteLimit": 2048000, "rowLimit": 10000 }, "finishTime": 1790341416693, "inputWidgets": {}, "nuid": "96224195-b4fd-4136-8a28-166ae9197489", "showTitle": true, "startTime": 1790341410864, "submitTime": 1790341410832, "tableResultSettingsMap": {}, "title": "Compare naive vs fixed totals" } }, "outputs": [], "source": [ "from pyspark.sql.functions import col\n", "\n", "naive_results = spark.sql(NAIVE).withColumnRenamed(\"total\", \"naive_total\")\n", "fixed_results = spark.sql(FIXED).withColumnRenamed(\"total\", \"fixed_total\")\n", "\n", "comparison = naive_results.join(fixed_results, \"c_mktsegment\")\n", "display(comparison.withColumn(\n", " \"difference\",\n", " col(\"naive_total\") - col(\"fixed_total\")\n", ").orderBy(\"c_mktsegment\"))\n" ] }, { "cell_type": "markdown", "metadata": { "application/vnd.databricks.v1+cell": { "cellMetadata": {}, "inputWidgets": {}, "nuid": "b9649453-ec23-4d38-8652-082f1eb8008d", "showTitle": false, "tableResultSettingsMap": {}, "title": "" } }, "source": [ "## 9. Your own slice\n", "\n", "Your salt picks 1 customer key in 64. Somebody else's notebook answers this with their own slice.\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": { "application/vnd.databricks.v1+cell": { "cellMetadata": { "byteLimit": 2048000, "rowLimit": 10000 }, "finishTime": 1790329584474, "inputWidgets": {}, "nuid": "c83850c8-c48b-4684-952c-70cd5c0d7733", "showTitle": false, "startTime": 1790329582175, "submitTime": 1790329582140, "tableResultSettingsMap": {}, "title": "" } }, "outputs": [], "source": [ "bucket = int(SALT, 16) % 64\n", "\n", "# TODO: total revenue for the orders whose customer key falls in your bucket\n", "# (`o_custkey % 64 = bucket`), joined the way your fixed query joins.\n", "probe = None\n", "\n", "print(f\"BRICKSTER:probe={probe}\")\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## 10. Check your run\n", "\n", "Two ways, same grader.\n", "\n", "**Export and upload.** *File > Export > IPYNB*, then drop it on the lab page.\n", "\n", "**Or straight from here.** Make a token on https://brickster.io/settings, put it in the `brickster`\n", "secret scope once per workspace, and run the last cell. Your Databricks credentials never leave\n", "this workspace.\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "%pip install brickster\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "import brickster\n", "\n", "brickster.submit(\"the-join-that-should-broadcast\")\n" ] } ], "metadata": { "application/vnd.databricks.v1+notebook": { "computePreferences": null, "dashboards": [], "environmentMetadata": { "base_environment": "", "environment_version": "5" }, "inputWidgetPreferences": null, "language": "python", "notebookMetadata": { "pythonIndentUnit": 4 }, "notebookName": "starter", "widgets": {} }, "kernelspec": { "display_name": "Python 3", "language": "python", "name": "python3" }, "language_info": { "name": "python" } }, "nbformat": 4, "nbformat_minor": 0 }