diff --git a/CHANGELOG.md b/CHANGELOG.md index c7414aa35..9d5362060 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -26,6 +26,9 @@ these packages. All five distributions share one version and one entry. `connections.toml` profile, `SF_PARTNER`, or the connector's own detection of Streamlit and notebooks. A connection or Snowpark session you open yourself is used as it is. +- A Databricks get-started notebook for the tabular model, + `examples/databricks/kumo_tabular_quickstart.ipynb`, which deploys it to a + serving endpoint and fills in a column of the AdventureWorks products table. - The relational engine publishes wheels for macOS (Apple silicon and Intel) and 64-bit Windows, and for CPython 3.14, alongside the `manylinux_2_28` x86-64 and aarch64 wheels it already built, so `structured-data-models-client[relational]` @@ -46,6 +49,9 @@ these packages. All five distributions share one version and one entry. ### Changed +- The refusal of a request over a Databricks serving endpoint's 16 MB limit + now asks for fewer context rows or fewer rows to score, covering both + directly supplied rows and requests built through predictive queries. - Every engine wheel is built on a machine of its own, native rather than emulated, and each is smoke-tested on the architecture it targets. The release no longer builds aarch64 under QEMU, which also takes a privileged diff --git a/docs/get-started/quickstart.md b/docs/get-started/quickstart.md index 750fffacd..c6ecb8c77 100644 --- a/docs/get-started/quickstart.md +++ b/docs/get-started/quickstart.md @@ -152,6 +152,16 @@ This needs `pip install 'structured-data-models-client[databricks-serving]'`. A answers one request at a time, so `fit()` is refused there; `predict()` sends the context with every call. +A `kumo-tabular` endpoint is reached the same way, with `model=KumoTabular()`. +The Databricks notebooks +[`kumo_relational_quickstart`](https://github.com/NVIDIA/structured-data-models-client/blob/main/examples/databricks/kumo_relational_quickstart.ipynb) and +[`kumo_tabular_quickstart`](https://github.com/NVIDIA/structured-data-models-client/blob/main/examples/databricks/kumo_tabular_quickstart.ipynb) +deploy each model to a serving endpoint and ask it questions about the +AdventureWorks sample data. + +The notebooks install `structured-data-models-client[databricks,databricks-serving]` +to include the warehouse connectors as well as the serving SDK. + ## A model in this process The model can also run in your own Python process, with no NIM at all. The diff --git a/docs/reference/limits.md b/docs/reference/limits.md index 1e83a93bb..309cf6c94 100644 --- a/docs/reference/limits.md +++ b/docs/reference/limits.md @@ -26,7 +26,7 @@ Both endpoints enforce sizes, and most of these are the relational one's. | entities one question gathers | relational | 100,000, in at most 100 calls | at `materialize()`, and this client's own: the endpoint is never asked for more than one call's worth | | calls a question asked in parts makes | relational | 200, one per candidate or step | at `materialize()`, as `sdm_client.decompose.MOST_PARTS` | | request body | relational | 30 MiB per call | at `materialize()` | -| request body | relational, through Databricks model serving | 16 MB per call, the gateway's own decimal limit | before sending, on the wrapped envelope | +| request body | relational and tabular, through Databricks model serving | 16 MB per call, the gateway's own decimal limit | before sending, on the wrapped envelope | | hops | relational | 6 | at `materialize()` | | tables | relational | 15 | against the graph the request carries, not the schema it was built from | diff --git a/examples/databricks/kumo_tabular_quickstart.ipynb b/examples/databricks/kumo_tabular_quickstart.ipynb new file mode 100644 index 000000000..0a9ec41f3 --- /dev/null +++ b/examples/databricks/kumo_tabular_quickstart.ipynb @@ -0,0 +1,676 @@ +{ + "cells": [ + { + "cell_type": "markdown", + "id": "18beb70fdc90345950ccb127fd06ff75", + "metadata": {}, + "source": [ + "# Get started with Kumo Tabular\n", + "\n", + "Kumo Tabular is a foundation model for a single table: it fills in a column from the other columns, with no feature engineering and no model trained for each task.\n", + "You give it rows where the answer is known and rows where it is not, and it predicts the missing answers from the known ones.\n", + "\n", + "This notebook deploys the model in your workspace, then asks three questions of Microsoft's AdventureWorks products table: which color a product has, what its list price is, and the color question again as a predictive query.\n", + "Each prediction is checked against the value the table records, and compared with a simple rule, such as always guessing the most common color." + ] + }, + { + "cell_type": "markdown", + "id": "94b63d84cfc7b541f6144d1d49a85e45", + "metadata": {}, + "source": [ + "## How Kumo Tabular works\n", + "\n", + "> **1. The input is one table.**\n", + "\n", + "The table holds numerical and categorical columns, such as a product's weight and its size.\n", + "The model reads no related tables and no time windows, so it needs no graph: rows in, answers out.\n", + "\n", + "> **2. Rows where the answer is known are the context.**\n", + "\n", + "The rows where the target column is filled in are the context examples.\n", + "The rows you want answered are the rows to score, and have the target column left out.\n", + "\n", + "> **3. A prediction is classification or regression.**\n", + "\n", + "A categorical target, such as a color, is a classification and returns a predicted class with a probability for each class.\n", + "A numerical target, such as a price, is a regression and returns a number.\n", + "\n", + "There is no training step: for each request, the client sends the context rows with the rows to score, and the model predicts from them." + ] + }, + { + "cell_type": "markdown", + "id": "1d2d1413ac076a95031bac81a0c15417", + "metadata": {}, + "source": [ + "## Before you start\n", + "\n", + "You need:\n", + "\n", + "- the Kumo Tabular model, in Unity Catalog or installed from Databricks Marketplace;\n", + "- permission to create a Model Serving endpoint, in a region with GPU Model Serving;\n", + "- a catalog and schema where you can create tables;\n", + "- serverless compute with environment 4 or later, or Databricks Runtime 17.3 LTS or later.\n", + "\n", + "The first deployment of the endpoint takes 20 to 40 minutes, then the rest of the notebook runs in a few minutes.\n", + "The endpoint bills for GPU time while it runs; it scales to zero when idle, and the last step shows how to delete it.\n", + "\n", + "Run each cell in order with Shift+Enter.\n", + "\n", + "**Let's get started!**" + ] + }, + { + "cell_type": "markdown", + "id": "7a0c878d48c1b6d0bebbe0a2d7bf40a4", + "metadata": {}, + "source": [ + "## Step 1. Install the client\n", + "\n", + "Install the client, the `structured-data-models-client` package, with its Databricks extras, then restart Python.\n", + "pip may print a red message about `protobuf` versions; it does not affect this notebook." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "5445169ae07af0aa529f51f1cb835515", + "metadata": {}, + "outputs": [], + "source": [ + "%pip install --quiet \"structured-data-models-client[databricks,databricks-serving]\"\n", + "dbutils.library.restartPython()" + ] + }, + { + "cell_type": "markdown", + "id": "c8a993bc41f4ddcb078426d5e83070f2", + "metadata": {}, + "source": [ + "## Step 2. Set your parameters\n", + "\n", + "Set these names, then run the cell.\n", + "`MODEL_NAME` is the Kumo Tabular model, in the catalog you installed it into, and `CATALOG` and `SCHEMA` are where the sample table goes." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "1e8e09e66cfe3045b176857b3dd28c3a", + "metadata": {}, + "outputs": [], + "source": [ + "from datetime import timedelta\n", + "\n", + "import pandas as pd\n", + "from databricks.sdk import WorkspaceClient\n", + "from databricks.sdk.errors import NotFound, OperationFailed\n", + "from databricks.sdk.service.serving import (\n", + " EndpointCoreConfigInput,\n", + " ServedEntityInput,\n", + " ServingModelWorkloadType,\n", + ")\n", + "from sdm_client import (\n", + " KumoTabular,\n", + " PredictiveQuery,\n", + " Request,\n", + " Schema,\n", + " SDMClient,\n", + ")\n", + "from sdm_connectors import quote_ident\n", + "\n", + "# The model, as .., and its version.\n", + "MODEL_NAME = 'kumo_tabular.models.kumo_tabular'\n", + "MODEL_VERSION = '1'\n", + "ENDPOINT_NAME = 'kumo-tabular'\n", + "# Where the sample table goes: a catalog and schema you can create tables in.\n", + "CATALOG = 'main'\n", + "SCHEMA = 'kumo_tabular_quickstart'\n", + "TABLE = 'adventureworks_products'\n", + "\n", + "\n", + "def quoted(*parts):\n", + " # A Unity Catalog name, each part quoted for SQL.\n", + " return '.'.join(quote_ident(part, char='`') for part in parts)\n", + "\n", + "\n", + "PRODUCTS = quoted(CATALOG, SCHEMA, TABLE)" + ] + }, + { + "cell_type": "markdown", + "id": "6c222fe7761b0460ac651bf4d49baea1", + "metadata": {}, + "source": [ + "## Step 3. Deploy the model\n", + "\n", + "The next cell creates a Model Serving endpoint for the model in your workspace, on one GPU (`GPU_MEDIUM`, size Small) that scales to zero when idle.\n", + "It does not wait for the deployment; Step 6 does.\n", + "If the endpoint already exists, the cell checks that it serves this model.\n", + "You can also create it with **Serve this model** on the model's page in Catalog Explorer." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "a133dc9b0e07efd33ac4da64235bcbdb", + "metadata": {}, + "outputs": [], + "source": [ + "workspace = WorkspaceClient()\n", + "try:\n", + " endpoint = workspace.serving_endpoints.get(ENDPOINT_NAME)\n", + "except NotFound:\n", + " workspace.serving_endpoints.create(\n", + " name=ENDPOINT_NAME,\n", + " config=EndpointCoreConfigInput(\n", + " name=ENDPOINT_NAME,\n", + " served_entities=[\n", + " ServedEntityInput(\n", + " entity_name=MODEL_NAME,\n", + " entity_version=MODEL_VERSION,\n", + " workload_type=ServingModelWorkloadType.GPU_MEDIUM,\n", + " workload_size='Small',\n", + " scale_to_zero_enabled=True,\n", + " )\n", + " ],\n", + " ),\n", + " )\n", + " print(\n", + " f'Creating the endpoint {ENDPOINT_NAME}; this takes 20 to 40 minutes.'\n", + " )\n", + "else:\n", + " update = endpoint.state.config_update if endpoint.state else None\n", + " if update and update.value in ('UPDATE_FAILED', 'UPDATE_CANCELED'):\n", + " # A failed pending configuration is not the active deployment.\n", + " config = endpoint.config\n", + " else:\n", + " config = endpoint.pending_config or endpoint.config\n", + " served = {\n", + " (entity.entity_name.lower(), entity.entity_version)\n", + " for entity in config.served_entities\n", + " }\n", + " if served != {(MODEL_NAME.lower(), MODEL_VERSION)}:\n", + " raise ValueError(\n", + " f'The endpoint {ENDPOINT_NAME} already exists and serves {sorted(served)}, '\n", + " f'not {MODEL_NAME} version {MODEL_VERSION}. Set ENDPOINT_NAME to another name.'\n", + " )\n", + " print(f'The endpoint {ENDPOINT_NAME} already exists and serves this model.')" + ] + }, + { + "cell_type": "markdown", + "id": "df62876d09335df77ab439e3935f6141", + "metadata": {}, + "source": [ + "## Step 4. Load the sample data\n", + "\n", + "The next cell downloads the AdventureWorks products from the [Microsoft SQL Server samples repository](https://github.com/microsoft/sql-server-samples/tree/master/samples/databases/adventure-works/oltp-install-script) at a fixed commit and saves them as a table in your schema.\n", + "It needs internet access, and skips the download if the table already exists with the expected columns.\n", + "\n", + "> Each user is responsible for checking the content of datasets and the applicable licenses and determining if suitable for the intended use.\n", + "> The repository's samples are provided under the [MIT License](https://github.com/microsoft/sql-server-samples/blob/master/license.txt)." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "1c9a61987b2e78e004b7da1d055a40b2", + "metadata": {}, + "outputs": [], + "source": [ + "URL = (\n", + " 'https://raw.githubusercontent.com/microsoft/sql-server-samples/'\n", + " 'b47eadc85277f72843647223cde53aa1b85ffcb8/samples/databases/'\n", + " 'adventure-works/oltp-install-script/Product.csv'\n", + ")\n", + "COLUMNS = (\n", + " 'ProductID Name ProductNumber MakeFlag FinishedGoodsFlag Color '\n", + " 'SafetyStockLevel ReorderPoint StandardCost ListPrice Size '\n", + " 'SizeUnitMeasureCode WeightUnitMeasureCode Weight DaysToManufacture '\n", + " 'ProductLine Class Style ProductSubcategoryID ProductModelID '\n", + " 'SellStartDate SellEndDate DiscontinuedDate rowguid ModifiedDate'\n", + ").split()\n", + "\n", + "if spark.catalog.tableExists(PRODUCTS):\n", + " if spark.table(PRODUCTS).columns != COLUMNS:\n", + " raise ValueError(\n", + " f'{PRODUCTS} already exists with other columns. '\n", + " 'Drop it, or set SCHEMA to another schema.'\n", + " )\n", + " print(f'The sample table is already in {CATALOG}.{SCHEMA}.')\n", + "else:\n", + " frame = pd.read_csv(\n", + " URL,\n", + " sep='\\t',\n", + " header=None,\n", + " names=COLUMNS,\n", + " dtype_backend='numpy_nullable',\n", + " )\n", + " for column in frame.columns:\n", + " if column.endswith('Date'):\n", + " frame[column] = pd.to_datetime(frame[column])\n", + " spark.sql(f'CREATE SCHEMA IF NOT EXISTS {quoted(CATALOG, SCHEMA)}')\n", + " spark.createDataFrame(frame).write.mode('ignore').saveAsTable(PRODUCTS)\n", + " print(f'Saved the sample table to {CATALOG}.{SCHEMA}.')" + ] + }, + { + "cell_type": "markdown", + "id": "fb44d873a5f8030a3e185da590015cb5", + "metadata": {}, + "source": [ + "## Step 5. Read the table and choose the features\n", + "\n", + "The model reads one table, so the table is read into the notebook: unlike the relational model, which samples linked tables where they are, there is nothing to sample here.\n", + "A table small enough to score is simply read.\n", + "\n", + "The model reads numerical and categorical columns.\n", + "The cell leaves out the rest: the product name, product number and `rowguid`, which name a row rather than describe it, and the dates, which are not numbers or categories.\n", + "It also leaves out `ProductID`, a row identifier, which Example 3 adds back as the key of a predictive query." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "7015e2d235fde236028420efe0c29b31", + "metadata": {}, + "outputs": [], + "source": [ + "products = (\n", + " spark.table(PRODUCTS).toPandas().sort_values('ProductID', ignore_index=True)\n", + ")\n", + "LEFT_OUT = [\n", + " 'ProductID',\n", + " 'rowguid',\n", + " 'ModifiedDate',\n", + " 'Name',\n", + " 'ProductNumber',\n", + " 'SellStartDate',\n", + " 'SellEndDate',\n", + " 'DiscontinuedDate',\n", + "]\n", + "features = products.drop(columns=LEFT_OUT)\n", + "print(f'{len(products)} products; features: {\", \".join(features.columns)}')\n", + "products.head()" + ] + }, + { + "cell_type": "markdown", + "id": "7bef8d59a20ee6a31c5b8f3e705a62a3", + "metadata": {}, + "source": [ + "## Step 6. Connect to the model\n", + "\n", + "The next cell waits until the endpoint is ready, then connects to it.\n", + "If the endpoint has scaled to zero, the first prediction takes a few extra minutes while it starts." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "3d8ef001398acc6e75952a6b3152b862", + "metadata": {}, + "outputs": [], + "source": [ + "update_error = None\n", + "try:\n", + " endpoint = (\n", + " workspace.serving_endpoints.wait_get_serving_endpoint_not_updating(\n", + " ENDPOINT_NAME, timeout=timedelta(hours=1)\n", + " )\n", + " )\n", + "except OperationFailed as error:\n", + " # A failed update can leave a healthy earlier deployment serving.\n", + " update_error = error\n", + " endpoint = workspace.serving_endpoints.get(ENDPOINT_NAME)\n", + "ready = endpoint.state.ready.value if endpoint.state.ready else 'unknown'\n", + "if ready != 'READY':\n", + " raise RuntimeError(\n", + " f'The endpoint {ENDPOINT_NAME} is {ready}. '\n", + " 'Open it under Serving to see why, then run this cell again.'\n", + " ) from update_error\n", + "# A pending update may have changed the model while we waited.\n", + "served = {\n", + " (entity.entity_name.lower(), entity.entity_version)\n", + " for entity in endpoint.config.served_entities\n", + "}\n", + "if served != {(MODEL_NAME.lower(), MODEL_VERSION)}:\n", + " raise ValueError(\n", + " f'The endpoint {ENDPOINT_NAME} now serves {sorted(served)}, '\n", + " f'not {MODEL_NAME} version {MODEL_VERSION}. '\n", + " 'Set ENDPOINT_NAME to another name and run from Step 3 again.'\n", + " ) from update_error\n", + "# Allow 15 minutes per request, for an endpoint that has to scale up from zero.\n", + "client = SDMClient.for_databricks_serving(ENDPOINT_NAME, timeout=900)\n", + "model = KumoTabular()\n", + "print(f'Connected to {ENDPOINT_NAME}.')" + ] + }, + { + "cell_type": "markdown", + "id": "cdcff7ffa76899086f96e1d9b857ac11", + "metadata": {}, + "source": [ + "## Step 7. Make predictions\n", + "\n", + "Each prediction takes two calls:\n", + "\n", + "1. `Request(context=..., predict=..., target=..., task=...)` holds the labelled rows, the rows to score, the column to predict, and the kind of question.\n", + "2. `client.predict(request, model=model)` sends it, and `to_pandas()` returns the result.\n", + "\n", + "`task` is required: it is `'classification'` or `'regression'`, and a column of whole numbers could be either.\n", + "\n", + "**Note:** AdventureWorks is sample data, and the results below are for demonstration.\n", + "Benchmark Kumo Tabular on data representative of your own use case before drawing conclusions about its performance." + ] + }, + { + "cell_type": "markdown", + "id": "e9b3e0cdbb9ae38d0faed7d135aa8b85", + "metadata": {}, + "source": [ + "### Example 1: Predict a product's color\n", + "\n", + "Which color does a product have?\n", + "\n", + "Some products have no color recorded, but only the products that do can be checked.\n", + "The cell sets aside a random 60 of those as the rows to score, with their color hidden, and gives the rest as context.\n", + "The result is one row per product to score, with the predicted class and a probability for each class." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "1304e32f1411c7445bf8c0f3098cac4d", + "metadata": {}, + "outputs": [], + "source": [ + "TARGET = 'Color'\n", + "known = features[features[TARGET].notna()].sample(frac=1, random_state=0)\n", + "held_out = known.head(60).reset_index(drop=True)\n", + "context = known.iloc[60:]\n", + "# Which rows are which, by their place in the table, for Example 3.\n", + "held_out_rows, context_rows = known.index[:60], known.index[60:]\n", + "\n", + "request = Request(\n", + " context=context,\n", + " predict=held_out.drop(columns=[TARGET]),\n", + " target=TARGET,\n", + " task='classification',\n", + ")\n", + "colors = client.predict(request, model=model).to_pandas()\n", + "colors.head()" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "6151e733e52d70acbff2706114823da7", + "metadata": {}, + "outputs": [], + "source": [ + "held_out['predicted'] = held_out.index.map(\n", + " colors.set_index('row_index')['prediction']\n", + ")\n", + "most_common = context[TARGET].mode()[0]\n", + "correct = (held_out['predicted'] == held_out[TARGET]).sum()\n", + "guessed = (held_out[TARGET] == most_common).sum()\n", + "print(\n", + " f'{correct} of {len(held_out)} colors predicted correctly; always '\n", + " f'guessing the most common color, {most_common}, gets {guessed} right.'\n", + ")\n", + "display(\n", + " held_out.loc[\n", + " held_out['predicted'] != held_out[TARGET], [TARGET, 'predicted']\n", + " ]\n", + ")" + ] + }, + { + "cell_type": "markdown", + "id": "626bd7b7d84310cb239b73a01b94dd1f", + "metadata": {}, + "source": [ + "How to interpret the result:\n", + "\n", + "1. The first table has one row per product, keyed by `row_index`, its position among the rows to score, with the predicted color in `prediction` and one probability column per color.\n", + "2. The printed line counts the correct predictions, and the right guesses of a rule that always picks the most common color.\n", + "3. The second table lists the products the model got wrong.\n", + "\n", + "The probabilities are the model's, not calibrated scores: read them as an ordering over the colors rather than as frequencies." + ] + }, + { + "cell_type": "markdown", + "id": "4e8f15fa3c864abe0944c5bd7eecf152", + "metadata": {}, + "source": [ + "### Example 2: Predict a product's list price\n", + "\n", + "The same table, a number instead of a label.\n", + "Changing the question means changing the target and the task; everything else is the same call.\n", + "Here the answer is a price, so the task is a regression, and the result has one `prediction` column and no probabilities.\n", + "\n", + "Only products with a list price above zero are used, because a price of zero marks a part that is never sold.\n", + "The cell sets aside a random 60 of them as the rows to score, with their price hidden, and compares the predictions with the recorded prices and with a rule that always guesses the median price.\n", + "The table also holds each product's standard cost, which tracks its list price closely, so this is an easy question; a column with less to give away is a harder one." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "5a4a6dcc8d052704213c857fe345cebd", + "metadata": {}, + "outputs": [], + "source": [ + "priced = features[features['ListPrice'] > 0].sample(frac=1, random_state=0)\n", + "to_price = priced.head(60).reset_index(drop=True)\n", + "\n", + "request = Request(\n", + " context=priced.iloc[60:],\n", + " predict=to_price.drop(columns=['ListPrice']),\n", + " target='ListPrice',\n", + " task='regression',\n", + ")\n", + "prices = client.predict(request, model=model).to_pandas()\n", + "\n", + "to_price['predicted'] = to_price.index.map(\n", + " prices.set_index('row_index')['prediction']\n", + ")\n", + "error = (to_price['predicted'] - to_price['ListPrice']).abs().mean()\n", + "median = priced.iloc[60:]['ListPrice'].median()\n", + "rule_error = (median - to_price['ListPrice']).abs().mean()\n", + "print(\n", + " f'Average error of the predicted price: ${error:,.2f}; always guessing '\n", + " f'the median price, ${median:,.2f}: ${rule_error:,.2f}.'\n", + ")\n", + "to_price[['ListPrice', 'predicted']].head(10)" + ] + }, + { + "cell_type": "markdown", + "id": "018d3ad2f0e002fe4b16eaf285e0e2e0", + "metadata": {}, + "source": [ + "How to interpret the result:\n", + "\n", + "1. The printed line compares the average difference between the predicted and recorded price with that of a rule that always guesses the median price.\n", + "2. The table shows the first ten products: the recorded price and the predicted one." + ] + }, + { + "cell_type": "markdown", + "id": "5bf68bea28f4377838271a6fff96ba84", + "metadata": {}, + "source": [ + "### Example 3: Ask the color question in PQL\n", + "\n", + "Example 1 built a `Request` by hand.\n", + "The Predictive Query Language says what to predict and which entity it is about, and gives the same question in one line:\n", + "\n", + "```\n", + "PREDICT products.Color ← the column to fill in\n", + "FOR EACH products.ProductID ← one prediction per product\n", + "```\n", + "\n", + "The schema comes from the rows already in the notebook, so this asks nothing more of the catalog.\n", + "`materialize(predict=...)` takes the rows to score, the same 60 products as Example 1, and builds the request.\n", + "`ProductID` is added back here because the schema needs a key.\n", + "Almost every other clause of the language, such as aggregations, time windows and `RANK`, is refused on one table, and the refusal says so." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "69de2c28ce1e8d8f058dc216a7b2a44b", + "metadata": {}, + "outputs": [], + "source": [ + "with_id = products.drop(columns=[c for c in LEFT_OUT if c != 'ProductID'])\n", + "ask = with_id.loc[held_out_rows].reset_index(drop=True)\n", + "schema = Schema.from_data(\n", + " {'products': with_id.loc[context_rows]},\n", + " primary_keys={'products': 'ProductID'},\n", + ")\n", + "request = PredictiveQuery(\n", + " f'PREDICT products.{TARGET} FOR EACH products.ProductID', schema\n", + ").materialize(predict=ask.drop(columns=[TARGET]), model=model)\n", + "pql_colors = client.predict(request, model=model).to_pandas()\n", + "\n", + "ask['predicted'] = ask['ProductID'].map(\n", + " pql_colors.set_index('id')['prediction']\n", + ")\n", + "print(\n", + " f'{(ask[\"predicted\"] == ask[TARGET]).sum()} of {len(ask)} colors predicted '\n", + " 'correctly.'\n", + ")\n", + "pql_colors.head()" + ] + }, + { + "cell_type": "markdown", + "id": "83812a9f0e71d8e85a42a7865b937a13", + "metadata": {}, + "source": [ + "How to interpret the result:\n", + "\n", + "1. The result is keyed by `ProductID`, in the `id` column, as well as by `row_index`.\n", + "2. The printed line counts the correct predictions for the same 60 products as Example 1, from the same context rows, so it matches Example 1's count." + ] + }, + { + "cell_type": "markdown", + "id": "35191b5835272b7849ed573c1de6294e", + "metadata": {}, + "source": [ + "## Step 8. Try it yourself\n", + "\n", + "The next cell asks the same price question as Example 2, with a number of context rows, `CONTEXT_ROWS`, that you can change.\n", + "It prints the average error of the predicted prices.\n", + "Change `CONTEXT_ROWS`, or uncomment the challenge line, and run it again." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "b3cfc087b258d3276f5ac18bbe7c1fb2", + "metadata": {}, + "outputs": [], + "source": [ + "CONTEXT_ROWS = 200\n", + "# Challenge: uncomment the line, run the cell again, and compare the error.\n", + "# CONTEXT_ROWS = 20 # how few examples can the model learn from?\n", + "\n", + "request = Request(\n", + " context=priced.iloc[60:].head(CONTEXT_ROWS),\n", + " predict=to_price.drop(columns=['ListPrice', 'predicted']),\n", + " target='ListPrice',\n", + " task='regression',\n", + ")\n", + "result = client.predict(request, model=model).to_pandas()\n", + "predicted = to_price.index.map(result.set_index('row_index')['prediction'])\n", + "print(\n", + " f'Average error with {CONTEXT_ROWS} context rows: '\n", + " f'${(predicted - to_price[\"ListPrice\"]).abs().mean():,.2f}.'\n", + ")" + ] + }, + { + "cell_type": "markdown", + "id": "6e837325e79e8c45fc589d24752670b6", + "metadata": {}, + "source": [ + "## Step 9. Use your own data\n", + "\n", + "For your own table, name the column to fill in and say which rows to score:\n", + "\n", + "```python\n", + "known = df[df[''].notna()]\n", + "unknown = df[df[''].isna()].drop(columns=[''])\n", + "request = Request(\n", + " context=known,\n", + " predict=unknown,\n", + " target='',\n", + " task='classification', # or 'regression'\n", + ")\n", + "result = client.predict(request, model=model).to_pandas()\n", + "```\n", + "\n", + "| Question | Target | Task |\n", + "|---|---|---|\n", + "| Which category does this item belong to? | a categorical column | `classification` |\n", + "| Will this customer churn? | a yes-or-no column | `classification` |\n", + "| What will this order be worth? | a numerical column | `regression` |\n", + "| What is this item's missing weight? | a numerical column | `regression` |\n", + "\n", + "For linked tables and questions about the future, such as who will stop ordering in the next 90 days, use Kumo Relational instead: it reads the links between tables and the time of each event.\n", + "\n", + "Tips:\n", + "\n", + "- Leave out random row IDs, free text, and dates: the model reads numerical and categorical columns.\n", + "- A request carries its rows in one call, up to 10,000 context rows and a limit of 16 MB on the request body; send fewer rows if the endpoint refuses it as too large.\n", + "- To keep the predictions, save them as a table: `spark.createDataFrame(result).write.saveAsTable('')`.\n", + "\n", + "Learn more:\n", + "\n", + "- [One table](https://github.com/NVIDIA/structured-data-models-client/blob/main/docs/get-started/quickstart.md#one-table) in the client's quickstart\n", + "- [Limits](https://github.com/NVIDIA/structured-data-models-client/blob/main/docs/reference/limits.md)" + ] + }, + { + "cell_type": "markdown", + "id": "3b725598852163af887b0ac980a1ed6c", + "metadata": {}, + "source": [ + "## Step 10. Clean up\n", + "\n", + "The endpoint bills for GPU time only while it runs.\n", + "To delete it, uncomment the last line and run the cell.\n", + "The sample table stays in your schema until you drop it." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "5948acae96bf3d2d978e9a2402f176fc", + "metadata": {}, + "outputs": [], + "source": [ + "client.close()\n", + "# workspace.serving_endpoints.delete(ENDPOINT_NAME)" + ] + } + ], + "metadata": { + "application/vnd.databricks.v1+notebook": { + "language": "python" + }, + "language_info": { + "name": "python" + } + }, + "nbformat": 4, + "nbformat_minor": 5 +} diff --git a/packages/structured-data-models-client/src/sdm_client/core/databricks_serving.py b/packages/structured-data-models-client/src/sdm_client/core/databricks_serving.py index c2d4d2896..78b60bdbf 100644 --- a/packages/structured-data-models-client/src/sdm_client/core/databricks_serving.py +++ b/packages/structured-data-models-client/src/sdm_client/core/databricks_serving.py @@ -213,8 +213,9 @@ def predict(self, payload: dict[str, Any]) -> dict[str, Any]: f'this request is {size:,} bytes once wrapped for serving ' f'endpoint {self._endpoint!r}, above the ' f'{self._max_request_bytes:,} byte limit it accepts. Send ' - f'fewer rows per call: a smaller num_context_samples, or ' - f'fewer entities to score', + f'fewer rows per call: fewer context rows (a smaller ' + f'num_context_samples, where the model takes one), or fewer ' + f'rows to score', code='INVALID_REQUEST', details={'bytes': size, 'limit': self._max_request_bytes}, ) diff --git a/packages/structured-data-models-client/tests/test_databricks_notebooks.py b/packages/structured-data-models-client/tests/test_databricks_notebooks.py index d2ef76445..4f1ce04bb 100644 --- a/packages/structured-data-models-client/tests/test_databricks_notebooks.py +++ b/packages/structured-data-models-client/tests/test_databricks_notebooks.py @@ -10,7 +10,10 @@ import ast import importlib import json +from datetime import timedelta from pathlib import Path +from types import SimpleNamespace +from unittest.mock import Mock import pytest @@ -37,6 +40,7 @@ def _python(source: str) -> str: def test_there_is_a_notebook_for_each_way_in() -> None: assert {path.name for path in NOTEBOOKS} >= { 'kumo_relational_quickstart.ipynb', + 'kumo_tabular_quickstart.ipynb', 'lakebase_relational.ipynb', } @@ -70,3 +74,227 @@ def test_every_name_imported_from_this_project_is_public( f"{notebook.name} imports {private}, which are not in their module's " f'__all__' ) + + +@pytest.mark.parametrize( + ('active_model', 'ready_model', 'ready_version', 'accepted'), + [ + ('catalog.schema.tabular', 'catalog.schema.tabular', '1', True), + ('catalog.schema.other', 'catalog.schema.tabular', '1', True), + ('catalog.schema.tabular', 'CATALOG.SCHEMA.TABULAR', '1', True), + ('catalog.schema.tabular', 'catalog.schema.other', '1', False), + ('catalog.schema.tabular', 'catalog.schema.tabular', '2', False), + ], +) +def test_tabular_notebook_checks_the_model_after_a_pending_update( + active_model: str, ready_model: str, ready_version: str, accepted: bool +) -> None: + cells = _cells(EXAMPLES / 'kumo_tabular_quickstart.ipynb', 'code') + deploy = next( + cell for cell in cells if 'workspace = WorkspaceClient()' in cell + ) + connect = next( + cell + for cell in cells + if 'wait_get_serving_endpoint_not_updating(' in cell + ) + + def config(name: str, version: str) -> SimpleNamespace: + return SimpleNamespace( + served_entities=[ + SimpleNamespace(entity_name=name, entity_version=version) + ] + ) + + workspace = Mock() + workspace.serving_endpoints.get.return_value = SimpleNamespace( + state=None, + config=config(active_model, '1'), + pending_config=config('catalog.schema.tabular', '1'), + ) + workspace.serving_endpoints.wait_get_serving_endpoint_not_updating.return_value = SimpleNamespace( + config=config(ready_model, ready_version), + state=SimpleNamespace(ready=SimpleNamespace(value='READY')), + ) + client_factory = Mock() + namespace = { + 'WorkspaceClient': lambda: workspace, + 'NotFound': LookupError, + 'MODEL_NAME': 'catalog.schema.tabular', + 'MODEL_VERSION': '1', + 'ENDPOINT_NAME': 'tabular-demo', + 'timedelta': timedelta, + 'SDMClient': SimpleNamespace(for_databricks_serving=client_factory), + 'KumoTabular': Mock(), + } + exec(deploy, namespace) + workspace.serving_endpoints.create.assert_not_called() + if accepted: + exec(connect, namespace) + client_factory.assert_called_once_with('tabular-demo', timeout=900) + else: + with pytest.raises( + ValueError, match='not catalog.schema.tabular version 1' + ): + exec(connect, namespace) + client_factory.assert_not_called() + + +@pytest.mark.parametrize( + ('active', 'pending', 'accepted'), + [ + ('tabular', None, True), + ('other', None, False), + ('tabular', 'other', False), + ], +) +def test_tabular_deploy_checks_active_and_pending_config( + active, pending, accepted +): + cell = next( + c + for c in _cells(EXAMPLES / 'kumo_tabular_quickstart.ipynb', 'code') + if 'workspace = WorkspaceClient()' in c + ) + + def config(model): + return SimpleNamespace( + served_entities=[ + SimpleNamespace(entity_name=model, entity_version='1') + ] + ) + + workspace = Mock() + workspace.serving_endpoints.get.return_value = SimpleNamespace( + state=None, + config=config(active), + pending_config=config(pending) if pending else None, + ) + namespace = dict( + WorkspaceClient=lambda: workspace, + NotFound=LookupError, + MODEL_NAME='tabular', + MODEL_VERSION='1', + ENDPOINT_NAME='demo', + ) + if accepted: + exec(cell, namespace) + else: + with pytest.raises(ValueError, match='already exists'): + exec(cell, namespace) + workspace.serving_endpoints.create.assert_not_called() + workspace.serving_endpoints.wait_get_serving_endpoint_not_updating.assert_not_called() + + +@pytest.mark.parametrize('failure', ['UPDATE_FAILED', 'UPDATE_CANCELED']) +@pytest.mark.parametrize( + ('ready', 'model', 'version', 'expected'), + [ + ('READY', 'tabular', '1', None), + ('NOT_READY', 'tabular', '1', RuntimeError), + ('READY', 'other', '1', ValueError), + ('READY', 'tabular', '2', ValueError), + ], +) +def test_tabular_failed_update_checks_active_deployment( + failure, ready, model, version, expected +): + class OperationFailed(Exception): + pass + + cell = next( + c + for c in _cells(EXAMPLES / 'kumo_tabular_quickstart.ipynb', 'code') + if 'wait_get_serving_endpoint_not_updating(' in c + ) + workspace = Mock() + error = OperationFailed(failure) + workspace.serving_endpoints.wait_get_serving_endpoint_not_updating.side_effect = error + workspace.serving_endpoints.get.return_value = SimpleNamespace( + state=SimpleNamespace( + ready=SimpleNamespace(value=ready), + config_update=SimpleNamespace(value=failure), + ), + pending_config=SimpleNamespace( + served_entities=[ + SimpleNamespace(entity_name='other', entity_version='2') + ] + ), + config=SimpleNamespace( + served_entities=[ + SimpleNamespace(entity_name=model, entity_version=version) + ] + ), + ) + factory = Mock() + namespace = dict( + workspace=workspace, + OperationFailed=OperationFailed, + ENDPOINT_NAME='demo', + MODEL_NAME='tabular', + MODEL_VERSION='1', + timedelta=timedelta, + SDMClient=SimpleNamespace(for_databricks_serving=factory), + KumoTabular=Mock(), + ) + if expected is None: + # Run the initial deployment check too: failed pending config must not + # prevent recovery of the matching active deployment. + namespace.update( + WorkspaceClient=lambda: workspace, NotFound=LookupError + ) + deploy = next( + c + for c in _cells(EXAMPLES / 'kumo_tabular_quickstart.ipynb', 'code') + if 'workspace = WorkspaceClient()' in c + ) + exec(deploy, namespace) + if expected: + with pytest.raises(expected) as caught: + exec(cell, namespace) + assert caught.value.__cause__ is error + factory.assert_not_called() + else: + exec(cell, namespace) + factory.assert_called_once_with('demo', timeout=900) + + +def test_tabular_sampling_is_independent_of_source_order(): + import pandas as pd + + cells = _cells(EXAMPLES / 'kumo_tabular_quickstart.ipynb', 'code') + prepare = next(c for c in cells if c.startswith('products =')) + classify = next(c for c in cells if c.startswith("TARGET = 'Color'")).split( + 'request = Request(' + )[0] + regress = next(c for c in cells if c.startswith('priced =')).split( + 'request = Request(' + )[0] + frame = pd.DataFrame( + { + 'ProductID': range(100), + 'Color': ['red', 'blue'] * 50, + 'ListPrice': range(1, 101), + } + ) + for name in [ + 'rowguid', + 'ModifiedDate', + 'Name', + 'ProductNumber', + 'SellStartDate', + 'SellEndDate', + 'DiscontinuedDate', + ]: + frame[name] = '' + results = [] + for source in [frame, frame.sample(frac=1, random_state=42)]: + spark = Mock() + spark.table.return_value.toPandas.return_value = source + namespace = dict(spark=spark, PRODUCTS='products') + exec(prepare, namespace) + exec(classify, namespace) + exec(regress, namespace) + results.append(namespace) + for name in ['held_out', 'context', 'to_price', 'priced']: + pd.testing.assert_frame_equal(results[0][name], results[1][name]) diff --git a/packages/structured-data-models-client/tests/test_databricks_serving.py b/packages/structured-data-models-client/tests/test_databricks_serving.py index 511995efa..2686ef9b9 100644 --- a/packages/structured-data-models-client/tests/test_databricks_serving.py +++ b/packages/structured-data-models-client/tests/test_databricks_serving.py @@ -439,6 +439,10 @@ def test_an_oversized_request_is_refused_before_it_is_sent(): assert error.value.details['limit'] == 200 assert error.value.details['bytes'] > 200 assert workspace.serving_endpoints.calls == [] + # The advice also covers directly supplied context and prediction rows, + # without assuming a predictive-query sampling parameter. + assert 'context rows' in str(error.value) + assert 'rows to score' in str(error.value) def test_the_size_counted_is_the_wrapped_envelope_not_the_payload():