{ "cells": [ { "cell_type": "markdown", "metadata": {}, "source": [ "![Redis](https://redis.io/wp-content/uploads/2024/04/Logotype.svg?auto=webp&quality=85,75&width=120)\n", "\n", "# Featureform On-Demand (Real-Time) Features\n", "\n", "In this recipe we define an **on-demand feature** in [**Featureform**](https://docs.featureform.com/) — a feature computed **at request time**, from the live request payload combined with a precomputed feature served from **Redis**.\n", "\n", "## Why on-demand features\n", "Some features can't be precomputed because they depend on data that only exists *at the moment of the request* — the amount of the transaction being scored right now, the user's current cart, the time of day. On-demand features let you express that last-mile computation **as a versioned Featureform resource** instead of scattered application code, so the logic that ran in training is the exact logic that runs in production.\n", "\n", "## What we'll build\n", "A fraud-style **risk ratio**: `incoming transaction amount ÷ the user's historical average`.\n", "- The **historical average** is a normal feature, precomputed by a SQL transformation and materialized to **Redis**.\n", "- The **incoming amount** is passed in at request time as a parameter.\n", "- An **on-demand feature** combines the two when you call `client.features(...)`." ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## The stack — all local, no Spark\n", "\n", "- **ClickHouse** — offline store; runs the SQL transformation for the historical average.\n", "- **Redis** — online store; serves that average at low latency.\n", "- **Featureform** coordinator.\n", "\n", "> ⚠️ **Needs local Docker; will not run on Colab or in CI.** The setup cells below launch everything this recipe needs — the Featureform coordinator (gRPC `localhost:7878`, dashboard `http://localhost`), ClickHouse, and Redis — starting each only if it isn't already running. You just need a running Docker daemon." ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Environment Setup\n", "\n", "### Install Python Dependencies" ] }, { "cell_type": "code", "execution_count": 1, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Note: you may need to restart the kernel to use updated packages.\n" ] } ], "source": [ "%pip install -q featureform redis clickhouse-connect numpy" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### Start Featureform, ClickHouse, and Redis\n", "\n", "`featureform deploy docker` starts the coordinator and is a no-op if it's already running. ClickHouse and Redis are (re)started fresh." ] }, { "cell_type": "code", "execution_count": 2, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "f6b00917c2e5cf1eb79317029d0d4b0a08eef1340a8a2dc5d09115b9ebfd40e6\n", "e91cec91410eda1f91d826a938fc6620f8ebad6d669a7cf845e21d27a2f6c81f\n", "Deploying Featureform on Docker\n", "Starting Docker deployment on Darwin 24.6.0\n", "Checking if featureform container exists...\n", "\tContainer featureform has status \"exited\"\n", "\tContainer featureform is stopped. Starting...\n", "\n", "Featureform is now running!\n", "To access the dashboard, visit http://localhost:80\n", "To apply definition files, run `featureform apply --host http://localhost:7878 --insecure`\n" ] } ], "source": [ "# NBVAL_SKIP\n", "# This notebook owns ClickHouse and Redis: start them fresh here and remove them\n", "# in the cleanup cell. The Featureform coordinator may be shared with / managed by\n", "# other tooling, so we only *ensure* it is up (deploy is a no-op if it already is)\n", "# and never remove it.\n", "#\n", "# No ports are published to the host: the coordinator reaches these containers\n", "# over the shared docker network by container IP, and the data-load cell talks to\n", "# ClickHouse via `docker exec`. Publishing ports would collide with other host\n", "# tools (e.g. Jupyter kernels grab ~9000, ClickHouse's native port).\n", "!docker rm -f clickhouse redis 2>/dev/null\n", "!docker run -d --name clickhouse -e CLICKHOUSE_SKIP_USER_SETUP=1 clickhouse/clickhouse-server:latest\n", "!docker run -d --name redis redis:8\n", "\n", "# Ensure the coordinator is running, then wait until it reports healthy — a\n", "# coordinator that is up but not yet ready fails the ClickHouse transformation.\n", "!featureform deploy docker\n", "!for i in $(seq 1 40); do [ \"$(docker inspect -f '{{.State.Health.Status}}' featureform 2>/dev/null)\" = healthy ] && echo \"coordinator healthy\" && break; sleep 3; done" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### Configure connections" ] }, { "cell_type": "code", "execution_count": 3, "id": "1523305d", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "clickhouse: 172.17.0.2 redis: 172.17.0.3\n" ] } ], "source": [ "import os\n", "import time\n", "import uuid\n", "import subprocess\n", "\n", "# Featureform coordinator (gRPC)\n", "FEATUREFORM_HOST = os.getenv(\"FEATUREFORM_HOST\", \"localhost:7878\")\n", "\n", "\n", "def _container_ip(name):\n", " \"\"\"IP of a container on the shared docker bridge network.\"\"\"\n", " return subprocess.run(\n", " [\"docker\", \"inspect\", \"-f\",\n", " \"{{range .NetworkSettings.Networks}}{{.IPAddress}}{{end}}\", name],\n", " capture_output=True, text=True,\n", " ).stdout.strip()\n", "\n", "\n", "# The coordinator shares the default docker bridge with the provider containers,\n", "# so it connects to them by container IP + internal port. Using the internal\n", "# network (instead of host.docker.internal + published ports) avoids collisions\n", "# with other host tools — notably Jupyter kernels, which grab ports around 9000,\n", "# ClickHouse's native port. Override with env vars if your setup differs.\n", "CLICKHOUSE_HOST = os.getenv(\"CLICKHOUSE_HOST\") or _container_ip(\"clickhouse\")\n", "CLICKHOUSE_NATIVE_PORT = int(os.getenv(\"CLICKHOUSE_NATIVE_PORT\", \"9000\"))\n", "CLICKHOUSE_USER = os.getenv(\"CLICKHOUSE_USER\", \"default\")\n", "CLICKHOUSE_PASSWORD = os.getenv(\"CLICKHOUSE_PASSWORD\", \"\")\n", "CLICKHOUSE_DATABASE = os.getenv(\"CLICKHOUSE_DATABASE\", \"default\")\n", "\n", "REDIS_HOST = os.getenv(\"REDIS_HOST\") or _container_ip(\"redis\")\n", "REDIS_PORT = int(os.getenv(\"REDIS_PORT\", \"6379\"))\n", "REDIS_PASSWORD = os.getenv(\"REDIS_PASSWORD\", \"\")\n", "\n", "print(\"clickhouse:\", CLICKHOUSE_HOST, \"redis:\", REDIS_HOST)\n", "\n", "# The coordinator's metadata is persistent (survives restarts), and Featureform\n", "# dedups equivalent resources and reuses their variant — which, against state a\n", "# previous run left behind, silently rewires a source to the wrong variant and\n", "# breaks the graph. Registering fresh *names* each run makes every run a clean,\n", "# self-contained apply that never collides with or reuses prior state.\n", "SUFFIX = uuid.uuid4().hex[:8]\n", "VARIANT = \"quickstart\"\n", "TX_NAME = \"transactions_\" + SUFFIX\n", "AVG_SRC_NAME = \"average_user_transaction_\" + SUFFIX\n", "AVG_FEAT_NAME = \"avg_transactions_\" + SUFFIX\n", "ONDEMAND_NAME = \"transaction_risk_ratio_\" + SUFFIX\n", "\n", "# Other systems may restart the coordinator at any moment, and it needs a few\n", "# seconds to warm up after launch. Wrap coordinator calls so transient gRPC\n", "# failures are retried instead of aborting the notebook.\n", "_TRANSIENT = (\"could not connect\", \"socket closed\", \"unavailable\", \"connection refused\")\n", "\n", "def with_ff_retry(fn, attempts=40, delay=5):\n", " for _i in range(attempts):\n", " try:\n", " return fn()\n", " except Exception as _e:\n", " if _i < attempts - 1 and any(s in str(_e).lower() for s in _TRANSIENT):\n", " time.sleep(delay)\n", " continue\n", " raise" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### Create a sample transactions table in ClickHouse\n", "\n", "We load a small transactions table so the average-transaction feature has data to aggregate." ] }, { "cell_type": "code", "execution_count": 4, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "rows: 500\n" ] } ], "source": [ "# NBVAL_SKIP\n", "import subprocess\n", "import numpy as np\n", "\n", "def ch(query, stdin=None):\n", " \"\"\"Run a ClickHouse query inside the container (no host ports involved).\"\"\"\n", " return subprocess.run(\n", " [\"docker\", \"exec\", \"-i\", \"clickhouse\", \"clickhouse-client\", \"--query\", query],\n", " input=stdin, text=True, capture_output=True,\n", " )\n", "\n", "# Wait for clickhouse-server inside the container to start accepting queries.\n", "for _ in range(30):\n", " if ch(\"SELECT 1\").returncode == 0:\n", " break\n", " time.sleep(2)\n", "\n", "ch(\"DROP TABLE IF EXISTS transactions\")\n", "ch(\"CREATE TABLE transactions (TransactionID String, CustomerID String, \"\n", " \"TransactionAmount Float64) ENGINE = MergeTree ORDER BY CustomerID\")\n", "\n", "rng = np.random.default_rng(42)\n", "csv = \"\".join(\n", " f\"T{i:05d},C{int(rng.integers(1000, 1050)):04d},{round(float(rng.gamma(2.0, 50.0)), 2)}\\n\"\n", " for i in range(500)\n", ")\n", "ch(\"INSERT INTO transactions FORMAT CSV\", stdin=csv)\n", "print(\"rows:\", ch(\"SELECT count() FROM transactions\").stdout.strip())" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Register providers, source, and the precomputed feature\n", "\n", "Standard setup: register ClickHouse + Redis, a SQL transformation for each user's average transaction, and a feature materialized to Redis. This is the value the on-demand feature will build on." ] }, { "cell_type": "code", "execution_count": 5, "metadata": {}, "outputs": [], "source": [ "import featureform as ff\n", "\n", "clickhouse = ff.register_clickhouse(\n", " name=\"clickhouse-quickstart\",\n", " description=\"ClickHouse offline store with transaction history\",\n", " host=CLICKHOUSE_HOST,\n", " port=CLICKHOUSE_NATIVE_PORT,\n", " user=CLICKHOUSE_USER,\n", " password=CLICKHOUSE_PASSWORD,\n", " database=CLICKHOUSE_DATABASE,\n", ")\n", "\n", "redis = ff.register_redis(\n", " name=\"redis-quickstart\",\n", " description=\"Redis online (inference) store\",\n", " host=REDIS_HOST,\n", " port=REDIS_PORT,\n", " password=REDIS_PASSWORD,\n", " db=0,\n", ")" ] }, { "cell_type": "code", "execution_count": 6, "metadata": {}, "outputs": [], "source": [ "transactions = clickhouse.register_table(\n", " name=\"transactions\", variant=\"quickstart\", table=\"transactions\",\n", ")\n", "\n", "@clickhouse.sql_transformation(variant=\"quickstart\")\n", "def average_user_transaction():\n", " return (\n", " \"SELECT CustomerID AS user_id, avg(TransactionAmount) AS avg_transaction_amt \"\n", " \"FROM {{transactions.quickstart}} GROUP BY CustomerID\"\n", " )\n", "\n", "@ff.entity\n", "class User:\n", " avg_transactions = ff.Feature(\n", " average_user_transaction[[\"user_id\", \"avg_transaction_amt\"]],\n", " variant=\"quickstart\",\n", " type=ff.Float32,\n", " inference_store=redis,\n", " )" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Define the on-demand feature\n", "\n", "An on-demand feature is a function decorated with `@ff.ondemand_feature`. Its signature is fixed: `(client, params, entities)`.\n", "- `client` — lets the function look up other (precomputed) features, e.g. from Redis.\n", "- `entities` — the entity keys passed at serving time.\n", "- `params` — arbitrary request-time inputs you supply per call.\n", "\n", "Here it fetches the user's stored average from Redis and divides the live amount by it. This function is registered and versioned like any feature — but it runs **client-side, at request time**." ] }, { "cell_type": "code", "execution_count": 7, "metadata": {}, "outputs": [], "source": [ "@ff.ondemand_feature(variant=\"quickstart\")\n", "def transaction_risk_ratio(client, params, entities):\n", " \"\"\"Live transaction amount relative to the user's historical average.\"\"\"\n", " avg = client.features([(\"avg_transactions\", \"quickstart\")], {\"user\": entities[\"user\"]})[0]\n", " incoming_amount = params[0]\n", " if not avg:\n", " return 0.0\n", " return float(incoming_amount) / float(avg)" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Apply\n", "\n", "`client.apply()` materializes the average into Redis and registers the on-demand feature definition." ] }, { "cell_type": "code", "execution_count": 8, "metadata": {}, "outputs": [ { "data": { "text/html": [ "
UserWarning: install \"ipywidgets\" for Jupyter support\n",
       "
\n" ], "text/plain": [ "UserWarning: install \"ipywidgets\" for Jupyter support\n" ] }, "metadata": {}, "output_type": "display_data" }, { "name": "stdout", "output_type": "stream", "text": [ "Applying Run: 2026-07-23t17-35-43\n", "Creating User default_owner \n", "Creating Provider clickhouse-quickstart \n", "Creating Provider redis-quickstart \n", "Creating Source Variant transactions quickstart\n", "Creating Source Variant average_user_transaction quickstart\n", "Creating Entity user \n", "Creating Feature Variant avg_transactions quickstart\n", "Creating Ondemand Feature transaction_risk_ratio quickstart\n", "\n" ] }, { "data": { "text/html": [ "
\n"
      ],
      "text/plain": []
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "# NBVAL_SKIP\n",
    "client = ff.Client(host=FEATUREFORM_HOST, insecure=True)\n",
    "with_ff_retry(lambda: client.apply(asynchronous=False, verbose=True))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Serve it — combine live input with the Redis-served average\n",
    "\n",
    "Pass the entity and the request-time `params` to `client.features()`. The same call would run behind a live fraud model: a ratio well above 1 means this transaction is large relative to the user's norm."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "No resources to apply\n",
      "user C1047: amount=92.00  avg=92.00  risk_ratio=[1.0]\n",
      "user C1047: amount=460.01  avg=92.00  risk_ratio=[5.0]\n"
     ]
    }
   ],
   "source": [
    "# NBVAL_SKIP\n",
    "def _serve():\n",
    "    user_id = client.dataframe(average_user_transaction)[\"user_id\"].iloc[0]\n",
    "    stored_avg = client.features([(\"avg_transactions\", \"quickstart\")], {\"user\": user_id})[0]\n",
    "    for incoming_amount in [stored_avg, stored_avg * 5]:\n",
    "        ratio = client.features(\n",
    "            [transaction_risk_ratio],\n",
    "            {\"user\": user_id},\n",
    "            params=[incoming_amount],\n",
    "        )\n",
    "        print(f\"user {user_id}: amount={incoming_amount:.2f}  avg={stored_avg:.2f}  risk_ratio={ratio}\")\n",
    "\n",
    "with_ff_retry(_serve)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Why this matters\n",
    "\n",
    "The division logic lives in **one versioned resource**, not duplicated across a training script and a serving service. Train on `transaction_risk_ratio` and you score production traffic with byte-for-byte the same computation — no training-serving skew, even for the real-time part."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Cleanup\n",
    "\n",
    "Stop and remove the containers when you're done."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "clickhouse\n",
      "redis\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "WARNING: All log messages before absl::InitializeLog() is called are written to STDERR\n",
      "I0000 00:00:1784853534.317493 3481186 chttp2_transport.cc:1353] ipv6:%5B::1%5D:7878: Got goaway [11] err=UNAVAILABLE:GOAWAY received; Error code: 11; Debug Text: too_many_pings {http2_error:11, grpc_status:14}\n",
      "E0000 00:00:1784853534.317794 3481186 chttp2_transport.cc:1385] ipv6:%5B::1%5D:7878: Received a GOAWAY with error code ENHANCE_YOUR_CALM and debug data equal to \"too_many_pings\". Current keepalive time (before throttling): 60000ms\n"
     ]
    }
   ],
   "source": [
    "# NBVAL_SKIP\n",
    "# Remove the ClickHouse and Redis containers this notebook started. The\n",
    "# coordinator is left alone in case it is shared with other tooling; stop it\n",
    "# yourself with `!featureform stop docker` if this notebook launched it.\n",
    "!docker rm -f clickhouse redis"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Learn more\n",
    "\n",
    "- [Featureform on-demand features](https://docs.featureform.com/)\n",
    "- [Featureform + Redis fraud detection recipe](./02_featureform_fraud_detection.ipynb)"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "redis-ai-res",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.11.9"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}