Recommend Products on Linux

Predict which products each customer will purchase in the next 90 days. This example trains a temporal link model and saves five ranked product candidates per customer. Evaluate the rankings against your recommendation objective before using them in a business workflow. For relationships without a future time window, use the optional static recommendation example. This guide uses the local Python API from the Linux setup. Do not combine its calls with legacy Kumo Studio examples.

Run the Complete Example

Complete Linux Machine Deployment Setup, then download the Linux examples bundle and prepare the sample CSV files. Use the four prepared AdventureWorks CSVs. Activate the installed Python environment and keep ADVENTUREWORKS_DATA set to the prepared CSV directory.

Run this command from the extracted bundle directory. Replace the two output paths with new, separate directories. Use local storage for scratch and retain results on storage whose lifetime meets your needs. Keep both directories separate from the inputs and from each other.

export KUMO_LINK_OUTPUT=/path/to/retained-results/adventureworks-links-01
python adventureworks_link.py \
  --input-dir "$ADVENTUREWORKS_DATA" \
  --scratch-dir /path/to/local-storage/adventureworks-links-scratch \
  --output-dir "$KUMO_LINK_OUTPUT" \
  --prediction-time 2025-06-30T00:00:00

The script builds the graph, generates historical labels from the event data, suggests a model plan, trains the model, and writes the model and recommendation outputs. Review the model plan and training budget before adapting it to production data. It saves the model under model/ and writes predictions.parquet with five product candidates for each customer.

A nonzero exit means the example did not complete. Preserve its error and existing outputs, correct the reported problem, and choose a new output directory before another attempt.

Understand the Prediction Task

The script uses this predictive query:

PREDICT LIST_DISTINCT(SalesOrderDetail.ProductID, 0, 90, days) RANK TOP 5 FOR EACH Customer.CustomerID

At each training time, LIST_DISTINCT collects the products that the customer purchases after that time and up to and including 90 days later. Repeated order details with the same product contribute one product identifier. The graph connects SalesOrderDetail.ProductID to Product.ProductID, so the target identifies product entities rather than ordinary category labels. The query’s task type is temporal_link_prediction. A label set can be empty or contain multiple product identifiers.

RANK TOP 5 asks for the top five products in the predictive query. The script also passes num_classes_to_return=5 to inference to set how many candidates it returns. Keep this explicit inference argument when adapting the example; do not assume the query sets the inference default.

The prediction time is June 30, 2025, at midnight UTC. Pass a datetime without timezone information; Kumo interprets it as UTC. The script sets this time when it generates the prediction table, then passes that table and the matching snapshot to inference.

Inspect the Saved Recommendations

After the script exits, read the output in a new Python process with the same KUMO_LINK_OUTPUT and ADVENTUREWORKS_DATA environment variables:

import os
from pathlib import Path
import numpy as np
import pandas as pd
output = Path(os.environ["KUMO_LINK_OUTPUT"]).expanduser().resolve()
inputs = Path(os.environ["ADVENTUREWORKS_DATA"]).expanduser().resolve()
predicted = pd.read_parquet(output / "predictions.parquet")
customers = pd.read_csv(inputs / "Customer.csv")
products = pd.read_csv(inputs / "Product.csv")
columns = ["ENTITY", "TIMESTAMP", "CLASS", "SCORE"]
missing = set(columns) - set(predicted.columns)
if missing:
    raise KeyError(f"Missing prediction columns: {sorted(missing)}")
assert predicted["ENTITY"].isin(customers["CustomerID"]).all()
assert predicted["CLASS"].isin(products["ProductID"]).all()
assert predicted.groupby(["ENTITY", "TIMESTAMP"]).size().eq(5).all()
assert np.isfinite(predicted["SCORE"].to_numpy()).all()
ranked = predicted.sort_values(
    ["ENTITY", "TIMESTAMP", "SCORE"], ascending=[True, True, False]
)
first_user = ranked["ENTITY"].iloc[0]
print("Rows:", len(predicted))
print(ranked.loc[ranked["ENTITY"].eq(first_user)])

Expect these columns:

Higher scores rank candidates ahead of lower scores for the same user and prediction time. Read CLASS as a product identifier and join it to Product.ProductID when you need product details. The output does not include a rank column, so the example sorts scores to display the recommendations in order. Scores can be negative. They are not guaranteed probabilities and do not need to sum to one. A returned product is a candidate recommendation, not an observed future purchase.

Handle Candidate and Time Errors

Use a positive num_classes_to_return for this task. Zero or a negative value raises a ValueError; use 5 to reproduce the example. The graph must retain two direct relationships from the order details. SalesOrderDetail.ProductID must point to the product primary key so the query can resolve its recommendation target. SalesOrderDetail.CustomerID must point to the customer primary key so the target events connect directly to the query population.

If the script rejects a timestamp with timezone information, supply the intended UTC time without the timezone suffix, as in the command above. Convert other local times to UTC before removing timezone information.

Use an appropriate evaluation dataset and training budget before relying on the rankings. See Making Predictions for prediction parameters and output retention, and Publish and Retrieve Models for model reload requirements.

Rank Relationships Without a Time Window

Use static link prediction to rank product relationships without asking when a purchase will occur. This is a separate example using the same four prepared CSV files. It does not use their timestamps or replace the temporal example above.

The static query has no time window:

PREDICT LIST_DISTINCT(SalesOrderDetail.ProductID) RANK TOP 5 FOR EACH Customer.CustomerID

Its task type is static_link_prediction. The graph must also omit the interaction time column. The adventureworks_static_link.py script keeps the same customer and product primary keys and all four relationships, but omits order timestamps.

sales_order_detail = Table.from_source_table(
    connector["SalesOrderDetail"],
    column_names=["SalesOrderDetailID", "SalesOrderID", "CustomerID", "ProductID", "OrderQty", "UnitPrice", "LineTotal"],
    primary_key="SalesOrderDetailID",
)
sales_order_detail = sales_order_detail.with_column("SalesOrderID", stype=Stype.ID)
sales_order_detail = sales_order_detail.with_column("CustomerID", stype=Stype.ID)
sales_order_detail = sales_order_detail.with_column("ProductID", stype=Stype.ID)

This excerpt belongs inside the script’s graph builder, where connector, Table, and Stype are already defined. Removing only the query’s time window while retaining the temporal graph raises an error requiring a time range. Keep separate scratch for the two graph definitions.

Set new scratch and output directories, then run the complete script from the extracted bundle:

export KUMO_STATIC_LINK_SCRATCH=/path/to/local-storage/adventureworks-static-link-scratch
export KUMO_STATIC_LINK_OUTPUT=/path/to/retained-results/adventureworks-static-link-01
python adventureworks_static_link.py fit \
  --input-dir "$ADVENTUREWORKS_DATA" \
  --scratch-dir "$KUMO_STATIC_LINK_SCRATCH" \
  --output-dir "$KUMO_STATIC_LINK_OUTPUT" \
  --num-classes-to-return 5

Each generated label set contains the customer’s distinct product identifiers across the source data, without time windows. Training, validation, and test partition individual relationships, so a user can appear in more than one split. The complete generated table is not itself the training split. Do not interpret the split as holding out users or future time periods.

The script publishes model/ and writes predictions.parquet. The static output has ENTITY, CLASS, and SCORE, with five distinct products per customer. It has no TIMESTAMP column and takes no prediction time. Generated intermediate tables contain an epoch timestamp placeholder, which is not a forecast date. Scores are ranking values, not guaranteed probabilities. This example does not guarantee exclusion of previously purchased products or recommendation quality.

Reload the Static Model

After the fit process exits, retain the complete trusted model directory. Provide compatible inputs with the same graph schema, relationships, and query fields. The prediction process can use a new local scratch directory and must use a new output directory. This mode loads the saved model and predicts without training:

export KUMO_STATIC_LINK_RELOAD=/path/to/retained-results/adventureworks-static-link-reload-01
python adventureworks_static_link.py predict \
  --input-dir "$ADVENTUREWORKS_DATA" \
  --scratch-dir "$KUMO_STATIC_LINK_SCRATCH" \
  --model-dir "$KUMO_STATIC_LINK_OUTPUT/model" \
  --output-dir "$KUMO_STATIC_LINK_RELOAD" \
  --num-classes-to-return 5

Use a separate new output directory each time and prevent overlapping processes against the same scratch. Load only a trusted model bundle. Reload requires a compatible snapshot and query. The model bundle does not contain source data or graph scratch. Updated data can be scored when it remains compatible.

Read the saved static recommendations in another Python process using the environment variables above:

import os
from pathlib import Path
import numpy as np
import pandas as pd
output = Path(os.environ["KUMO_STATIC_LINK_RELOAD"]).expanduser().resolve()
inputs = Path(os.environ["ADVENTUREWORKS_DATA"]).expanduser().resolve()
predicted = pd.read_parquet(output / "predictions.parquet")
customers = pd.read_csv(inputs / "Customer.csv")
products = pd.read_csv(inputs / "Product.csv")
columns = ["ENTITY", "CLASS", "SCORE"]
missing = set(columns) - set(predicted.columns)
if missing:
    raise KeyError(f"Missing prediction columns: {sorted(missing)}")
assert predicted["ENTITY"].isin(customers["CustomerID"]).all()
assert predicted["CLASS"].isin(products["ProductID"]).all()
assert predicted.groupby("ENTITY").size().eq(5).all()
assert np.isfinite(predicted["SCORE"].to_numpy()).all()
assert all(
    group["SCORE"].is_monotonic_decreasing
    for _, group in predicted.groupby("ENTITY")
)
first_user = predicted["ENTITY"].iloc[0]
print("Rows:", len(predicted))
print(predicted.loc[predicted["ENTITY"].eq(first_user)])