{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "40970080",
   "metadata": {},
   "source": [
    "![Redis](https://redis.io/wp-content/uploads/2024/04/Logotype.svg?auto=webp&quality=85,75&width=120)\n",
    "\n",
    "# Featureform Transformations & Feature Lineage\n",
    "\n",
    "In this recipe we build features with **SQL transformations** in [**Featureform**](https://docs.featureform.com/), and follow the **lineage** Featureform records from raw data all the way to a served feature.\n",
    "\n",
    "## Why transformations and lineage matter\n",
    "A production feature is rarely a raw column — it's the result of transformations that clean and aggregate the raw events. Two things go wrong without a feature store:\n",
    "- **Nobody can tell how a value was produced.** When a model misbehaves you need to trace a feature back to its source. Ad-hoc scripts don't record that path.\n",
    "- **Steps get duplicated and drift.** The \"clean transactions\" logic gets rewritten slightly differently by every downstream job.\n",
    "\n",
    "Featureform fixes both by making every transformation a **named, versioned resource** whose input is another named resource. That dependency graph *is* the lineage: `raw source → transformation → feature`, recorded and visualizable, with each step defined exactly once.\n",
    "\n",
    "## What we'll build\n",
    "Two SQL transformations over a transactions dataset, each derived from the same raw source:\n",
    "1. **`clean_transactions`** — filters out invalid rows from the raw source.\n",
    "2. **`avg_user_transaction`** — aggregates each user's valid transactions into an average and a count.\n",
    "\n",
    "Each transformation is a named node whose input is the `transactions` source, so Featureform records a lineage edge `transactions → <transformation>` for each. We then define an `avg_transaction_amt` feature from `avg_user_transaction`, a label from `clean_transactions`, register a training set, `apply()` everything, and walk each node — seeing the exact DataFrame Featureform computed."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "4482a5d6",
   "metadata": {},
   "source": [
    "## The stack — no Spark, no cloud\n",
    "\n",
    "Featureform separates **compute/offline** (where transformations run) from the **online store** (where features are served). Transformations here are **SQL**, which run directly in a SQL offline store — so there's no Spark cluster and no object storage to stand up. This recipe uses:\n",
    "- **ClickHouse** as the offline store — a columnar SQL database that runs the transformations. One local Docker container.\n",
    "- **Redis** as the online store, for low-latency serving of the finished feature.\n",
    "\n",
    "> ℹ️ **Transformations are SQL, not pandas here.** Featureform's pandas (`df_transformation`) support requires a Spark or Kubernetes provider. SQL transformations cover the same clean → aggregate → serve pipeline with none of that infrastructure.\n",
    "\n",
    "> ⚠️ **This notebook needs local Docker and will not run on Colab or in CI.** The cells below start all three pieces for you:\n",
    "> 1. A **ClickHouse** container (offline store).\n",
    "> 2. A **Redis** container (online store).\n",
    "> 3. The **Featureform** coordinator (gRPC on `localhost:7878`, dashboard on `http://localhost`) — started with `featureform deploy docker` after the pip install."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "2be36500",
   "metadata": {},
   "source": [
    "### Start ClickHouse and Redis\n",
    "\n",
    "This notebook launches its own containers on a private Docker network (`ff-net`) so the Featureform coordinator can reach them **by container name** — no reliance on host ports, which other processes may already be using. The containers are named `ff-clickhouse` / `ff-redis` (project-scoped, so they won't collide with anything else on your machine). Only ClickHouse's HTTP port is published to the host — on `18123` — so the cells below can load data; the native port and Redis stay inside `ff-net`, reached only by the coordinator."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "ad3fa2c6",
   "metadata": {
    "execution": {
     "iopub.execute_input": "2026-07-23T23:10:17.191416Z",
     "iopub.status.busy": "2026-07-23T23:10:17.191178Z",
     "iopub.status.idle": "2026-07-23T23:10:27.077023Z",
     "shell.execute_reply": "2026-07-23T23:10:27.075497Z"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "b499f2dd3bdcf1cb2211f7e82a7713f61a95b2ee25e24f541af1a607f5b6f783\n",
      "0c6f537968b1eeaf1414c60de80d3f2981efd10288e4302c7c07011e5af88b39\n",
      "5dd6a2ed98bd9c7519dfe4705eb2985d722a00b8649ecada48def33d693b4a60\n"
     ]
    }
   ],
   "source": [
    "# NBVAL_SKIP\n",
    "# Launch this notebook's own containers on a private Docker network (ff-net) so the\n",
    "# Featureform coordinator reaches them by name — no host-port collisions. Names are\n",
    "# project-scoped (ff-clickhouse/ff-redis) so they won't clash with other containers on\n",
    "# your machine. Only ClickHouse's HTTP port is published (on 18123, to avoid the common\n",
    "# 8123) so this notebook can load data; the native port and Redis stay inside ff-net.\n",
    "# ClickHouse is pinned to 24.10 — the native-protocol version the coordinator speaks.\n",
    "!docker rm -f ff-clickhouse ff-redis 2>/dev/null\n",
    "!docker network create ff-net 2>/dev/null || true\n",
    "!docker run -d --network ff-net --name ff-clickhouse -p 18123:8123 -e CLICKHOUSE_PASSWORD=featureform clickhouse/clickhouse-server:24.10\n",
    "!docker run -d --network ff-net --name ff-redis redis:8"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "7990ed95",
   "metadata": {},
   "source": [
    "## Environment Setup\n",
    "\n",
    "### Install Python Dependencies"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "c18c2952",
   "metadata": {
    "execution": {
     "iopub.execute_input": "2026-07-23T23:10:27.079832Z",
     "iopub.status.busy": "2026-07-23T23:10:27.079650Z",
     "iopub.status.idle": "2026-07-23T23:10:30.052383Z",
     "shell.execute_reply": "2026-07-23T23:10:30.051900Z"
    }
   },
   "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 pandas"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "4941db3e",
   "metadata": {
    "execution": {
     "iopub.execute_input": "2026-07-23T23:10:30.054406Z",
     "iopub.status.busy": "2026-07-23T23:10:30.054240Z",
     "iopub.status.idle": "2026-07-23T23:10:48.682578Z",
     "shell.execute_reply": "2026-07-23T23:10:48.681827Z"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Deploying Featureform on Docker\n",
      "Starting Docker deployment on Darwin 24.6.0\n",
      "Checking if featureform container exists...\n",
      "\tContainer featureform not found. Creating new container...\n",
      "\t'featureform' container started\n",
      "\n",
      "Featureform is now running!\n",
      "To access the dashboard, visit http://localhost:80\n",
      "To apply definition files, run `featureform apply <file.py> --host http://localhost:7878 --insecure`\n",
      "coordinator attached to ff-net\n"
     ]
    }
   ],
   "source": [
    "# NBVAL_SKIP\n",
    "import sys\n",
    "# Start the Featureform coordinator (gRPC on localhost:7878, dashboard on http://localhost).\n",
    "# Invoke via the kernel's own interpreter so it works even if the `featureform` console\n",
    "# script isn't on PATH (common in Jupyter/VSCode after %pip install). Then attach the\n",
    "# coordinator to ff-net, retrying until confirmed, so it resolves ClickHouse/Redis by name.\n",
    "!{sys.executable} -m featureform deploy docker\n",
    "!for i in $(seq 10); do docker network connect ff-net featureform 2>/dev/null; docker inspect featureform --format '{{range $k,$v := .NetworkSettings.Networks}}{{$k}} {{end}}' 2>/dev/null | grep -q ff-net && echo \"coordinator attached to ff-net\" && break; sleep 1; done"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "34ea9d78",
   "metadata": {},
   "source": [
    "### Configure connections\n",
    "\n",
    "Connection values are driven by environment variables so you can point the notebook at your own instances. The defaults are the **container names** on the shared `ff-net` network (`clickhouse`, `redis`) — how the coordinator container resolves the providers. (Data is loaded into ClickHouse from this notebook over the published HTTP port `8123` on `localhost`, in the cell further below.)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "400f7178",
   "metadata": {
    "execution": {
     "iopub.execute_input": "2026-07-23T23:10:48.685535Z",
     "iopub.status.busy": "2026-07-23T23:10:48.685285Z",
     "iopub.status.idle": "2026-07-23T23:10:48.689887Z",
     "shell.execute_reply": "2026-07-23T23:10:48.689536Z"
    }
   },
   "outputs": [],
   "source": [
    "import os\n",
    "\n",
    "# Featureform coordinator (gRPC)\n",
    "FEATUREFORM_HOST = os.getenv(\"FEATUREFORM_HOST\", \"localhost:7878\")\n",
    "\n",
    "# The coordinator reaches the providers by container name over the shared ff-net network.\n",
    "# ClickHouse offline store (recent images require a password for network access)\n",
    "CLICKHOUSE_HOST = os.getenv(\"CLICKHOUSE_HOST\", \"ff-clickhouse\")\n",
    "CLICKHOUSE_NATIVE_PORT = int(os.getenv(\"CLICKHOUSE_NATIVE_PORT\", \"9000\"))\n",
    "CLICKHOUSE_HTTP_PORT = int(os.getenv(\"CLICKHOUSE_HTTP_PORT\", \"18123\"))  # published to host for the data-load cell\n",
    "CLICKHOUSE_USER = os.getenv(\"CLICKHOUSE_USER\", \"default\")\n",
    "CLICKHOUSE_PASSWORD = os.getenv(\"CLICKHOUSE_PASSWORD\", \"featureform\")\n",
    "CLICKHOUSE_DATABASE = os.getenv(\"CLICKHOUSE_DATABASE\", \"default\")\n",
    "\n",
    "# Redis online store\n",
    "REDIS_HOST = os.getenv(\"REDIS_HOST\", \"ff-redis\")\n",
    "REDIS_PORT = int(os.getenv(\"REDIS_PORT\", \"6379\"))\n",
    "REDIS_PASSWORD = os.getenv(\"REDIS_PASSWORD\", \"\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "82a4158c",
   "metadata": {},
   "source": [
    "### Create a sample table in ClickHouse\n",
    "\n",
    "So the notebook is self-contained, we create a `transactions` table and load a small, intentionally *messy* dataset (some invalid negative amounts) so the first transformation has something to clean. In a real deployment this table would already exist. We connect over ClickHouse's published HTTP port `18123` from here."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "4b6e3a08",
   "metadata": {
    "execution": {
     "iopub.execute_input": "2026-07-23T23:10:48.691703Z",
     "iopub.status.busy": "2026-07-23T23:10:48.691552Z",
     "iopub.status.idle": "2026-07-23T23:10:49.359371Z",
     "shell.execute_reply": "2026-07-23T23:10:49.359119Z"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "rows in ClickHouse: 500\n"
     ]
    }
   ],
   "source": [
    "# NBVAL_SKIP\n",
    "import time\n",
    "import clickhouse_connect\n",
    "import numpy as np\n",
    "\n",
    "# ClickHouse may still be starting up — retry until it accepts connections.\n",
    "for _ in range(30):\n",
    "    try:\n",
    "        ch = clickhouse_connect.get_client(\n",
    "            host=\"localhost\", port=CLICKHOUSE_HTTP_PORT,\n",
    "            username=CLICKHOUSE_USER, password=CLICKHOUSE_PASSWORD,\n",
    "        )\n",
    "        ch.command(\"SELECT 1\")\n",
    "        break\n",
    "    except Exception:\n",
    "        time.sleep(1)\n",
    "else:\n",
    "    raise RuntimeError(\"ClickHouse not reachable on localhost:8123 — is the container running?\")\n",
    "\n",
    "ch.command(\"DROP TABLE IF EXISTS transactions\")\n",
    "ch.command(\n",
    "    \"\"\"\n",
    "    CREATE TABLE transactions (\n",
    "        TransactionID String,\n",
    "        CustomerID String,\n",
    "        TransactionAmount Float64,\n",
    "        IsFraud Bool\n",
    "    ) ENGINE = MergeTree ORDER BY CustomerID\n",
    "    \"\"\"\n",
    ")\n",
    "\n",
    "rng = np.random.default_rng(42)\n",
    "n = 500\n",
    "rows = []\n",
    "for i in range(n):\n",
    "    amount = round(float(rng.gamma(2.0, 50.0)), 2)\n",
    "    if i % 150 == 0:\n",
    "        amount = -1.0   # a few invalid amounts for the cleaning step to drop\n",
    "    rows.append([\n",
    "        f\"T{i:05d}\",\n",
    "        f\"C{int(rng.integers(1000, 1050)):04d}\",\n",
    "        amount,\n",
    "        bool(rng.integers(0, 2)),\n",
    "    ])\n",
    "\n",
    "ch.insert(\"transactions\", rows,\n",
    "          column_names=[\"TransactionID\", \"CustomerID\", \"TransactionAmount\", \"IsFraud\"])\n",
    "print(\"rows in ClickHouse:\", ch.command(\"SELECT count() FROM transactions\"))"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "4d102dc7",
   "metadata": {},
   "source": [
    "## Register the providers\n",
    "\n",
    "We register the **ClickHouse** offline store and the **Redis** online store. Registering a provider just tells Featureform how to reach it — no data moves yet."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "id": "7e43fd02",
   "metadata": {
    "execution": {
     "iopub.execute_input": "2026-07-23T23:10:49.360834Z",
     "iopub.status.busy": "2026-07-23T23:10:49.360704Z",
     "iopub.status.idle": "2026-07-23T23:10:49.751733Z",
     "shell.execute_reply": "2026-07-23T23:10:49.751094Z"
    }
   },
   "outputs": [],
   "source": [
    "import featureform as ff\n",
    "\n",
    "clickhouse = ff.register_clickhouse(\n",
    "    name=\"clickhouse-quickstart\",\n",
    "    description=\"ClickHouse offline store (runs the SQL transformations)\",\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": "markdown",
   "id": "cebda6c4",
   "metadata": {},
   "source": [
    "## Register the raw source\n",
    "\n",
    "We point Featureform at the ClickHouse `transactions` table. This becomes the **root node** of our lineage graph — the source that transformations build on."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "id": "4d63858c",
   "metadata": {
    "execution": {
     "iopub.execute_input": "2026-07-23T23:10:49.753526Z",
     "iopub.status.busy": "2026-07-23T23:10:49.753307Z",
     "iopub.status.idle": "2026-07-23T23:10:49.755417Z",
     "shell.execute_reply": "2026-07-23T23:10:49.755173Z"
    }
   },
   "outputs": [],
   "source": [
    "transactions = clickhouse.register_table(\n",
    "    name=\"transactions\",\n",
    "    variant=\"quickstart\",\n",
    "    table=\"transactions\",  # the table name in ClickHouse\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "4b567f3b",
   "metadata": {},
   "source": [
    "## Step 1 — a cleaning transformation\n",
    "\n",
    "A `sql_transformation` is a plain function that returns a SQL string. The `{{transactions.quickstart}}` placeholder references the source we just registered. This query runs **in ClickHouse**, and its result becomes a new, named source. Here we drop the invalid (non-positive) amounts."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "id": "e279314b",
   "metadata": {
    "execution": {
     "iopub.execute_input": "2026-07-23T23:10:49.756733Z",
     "iopub.status.busy": "2026-07-23T23:10:49.756625Z",
     "iopub.status.idle": "2026-07-23T23:10:49.759356Z",
     "shell.execute_reply": "2026-07-23T23:10:49.759035Z"
    }
   },
   "outputs": [],
   "source": [
    "@clickhouse.sql_transformation(variant=\"quickstart\")\n",
    "def clean_transactions():\n",
    "    \"\"\"Keep only valid transactions.\"\"\"\n",
    "    return (\n",
    "        \"SELECT CustomerID, TransactionAmount, IsFraud \"\n",
    "        \"FROM {{transactions.quickstart}} WHERE TransactionAmount > 0\"\n",
    "    )"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "63a1aaea",
   "metadata": {},
   "source": [
    "## Step 2 — a second transformation off the same source\n",
    "\n",
    "`avg_user_transaction` reads the **same `{{transactions.quickstart}}` source** and aggregates each user's valid transactions into an average and a count. It's a *sibling* of `clean_transactions`, not chained onto it: both are named, versioned nodes rooted in the raw source, so Featureform records a lineage edge from `transactions` to each.\n",
    "\n",
    "> ℹ️ Chaining one transformation directly onto another's output (`FROM {{clean_transactions.quickstart}}`) requires a Spark or Snowflake offline store. On a SQL offline store like ClickHouse, transformations read registered sources, so both transformations here derive from `transactions`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "id": "2a519281",
   "metadata": {
    "execution": {
     "iopub.execute_input": "2026-07-23T23:10:49.760988Z",
     "iopub.status.busy": "2026-07-23T23:10:49.760879Z",
     "iopub.status.idle": "2026-07-23T23:10:49.763088Z",
     "shell.execute_reply": "2026-07-23T23:10:49.762825Z"
    }
   },
   "outputs": [],
   "source": [
    "@clickhouse.sql_transformation(variant=\"quickstart\")\n",
    "def avg_user_transaction():\n",
    "    \"\"\"Average transaction amount and count per user, over the valid transactions.\"\"\"\n",
    "    return (\n",
    "        \"SELECT CustomerID AS user_id, \"\n",
    "        \"avg(TransactionAmount) AS avg_transaction_amt, \"\n",
    "        \"count(*) AS transaction_count \"\n",
    "        \"FROM {{transactions.quickstart}} WHERE TransactionAmount > 0 GROUP BY CustomerID\"\n",
    "    )"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "7e3779df",
   "metadata": {},
   "source": [
    "## Define the entity, feature, and label\n",
    "\n",
    "`@ff.entity` groups resources keyed by a **user**. The feature is sourced from the `avg_user_transaction` transformation and materialized to Redis for serving. The label (`IsFraud`) comes from the `clean_transactions` transformation and stays offline — it's only used to build training sets.\n",
    "\n",
    "Feature and label draw from **two different transformations rooted in the same raw source**, so they share a common lineage back to `transactions`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "id": "0c08d09e",
   "metadata": {
    "execution": {
     "iopub.execute_input": "2026-07-23T23:10:49.764364Z",
     "iopub.status.busy": "2026-07-23T23:10:49.764269Z",
     "iopub.status.idle": "2026-07-23T23:10:49.766864Z",
     "shell.execute_reply": "2026-07-23T23:10:49.766624Z"
    }
   },
   "outputs": [],
   "source": [
    "@ff.entity\n",
    "class User:\n",
    "    avg_transactions = ff.Feature(\n",
    "        avg_user_transaction[[\"user_id\", \"avg_transaction_amt\"]],\n",
    "        variant=\"quickstart\",\n",
    "        type=ff.Float32,\n",
    "        inference_store=redis,\n",
    "    )\n",
    "    fraudulent = ff.Label(\n",
    "        clean_transactions[[\"CustomerID\", \"IsFraud\"]],\n",
    "        variant=\"quickstart\",\n",
    "        type=ff.Bool,\n",
    "    )"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "cd6aec4c",
   "metadata": {},
   "source": [
    "## Register a training set\n",
    "\n",
    "A training set joins the feature(s) to the label on the entity key — built from the same definitions that serve online, so there's no training-serving skew."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "id": "f841f44a",
   "metadata": {
    "execution": {
     "iopub.execute_input": "2026-07-23T23:10:49.768149Z",
     "iopub.status.busy": "2026-07-23T23:10:49.768022Z",
     "iopub.status.idle": "2026-07-23T23:10:49.771483Z",
     "shell.execute_reply": "2026-07-23T23:10:49.771237Z"
    }
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "TrainingSetVariant(name='fraud_training', owner='default_owner', label=('fraudulent', 'quickstart'), features=[('avg_transactions', 'quickstart')], description='', variant='quickstart', feature_lags=[], tags=[], properties={}, created=None, schedule='', schedule_obj=None, provider='', status='NO_STATUS', error=None, server_status=None, resource_snowflake_config=None, type=<TrainingSetType.DYNAMIC: 1>)"
      ]
     },
     "execution_count": 11,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "ff.register_training_set(\n",
    "    \"fraud_training\",\n",
    "    variant=\"quickstart\",\n",
    "    label=(\"fraudulent\", \"quickstart\"),\n",
    "    features=[(\"avg_transactions\", \"quickstart\")],\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "7b92d249",
   "metadata": {},
   "source": [
    "## Apply the definitions\n",
    "\n",
    "`client.apply()` sends everything to the coordinator and runs the pipeline: ClickHouse executes both transformations against the `transactions` source, then the feature is materialized into Redis. `asynchronous=False` blocks until it finishes."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "id": "5c2aacc5",
   "metadata": {
    "execution": {
     "iopub.execute_input": "2026-07-23T23:10:49.772642Z",
     "iopub.status.busy": "2026-07-23T23:10:49.772568Z",
     "iopub.status.idle": "2026-07-23T23:10:58.237370Z",
     "shell.execute_reply": "2026-07-23T23:10:58.236195Z"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Applying Run: 2026-07-24t15-39-09\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 clean_transactions quickstart\n",
      "Creating Source Variant avg_user_transaction quickstart\n",
      "Creating Entity user \n",
      "Creating Feature Variant avg_transactions quickstart\n",
      "Creating Label Variant fraudulent quickstart\n",
      "Creating Trainingset Variant fraud_training quickstart\n",
      "\n"
     ]
    },
    {
     "data": {
      "text/html": [
       "<pre style=\"white-space:pre;overflow-x:auto;line-height:normal;font-family:Menlo,'DejaVu Sans Mono',consolas,'Courier New',monospace\">UserWarning: install \"ipywidgets\" for Jupyter support\n",
       "</pre>\n"
      ],
      "text/plain": [
       "UserWarning: install \"ipywidgets\" for Jupyter support\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/html": [
       "<pre style=\"white-space:pre;overflow-x:auto;line-height:normal;font-family:Menlo,'DejaVu Sans Mono',consolas,'Courier New',monospace\"></pre>\n"
      ],
      "text/plain": []
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "# NBVAL_SKIP\n",
    "client = ff.Client(host=FEATUREFORM_HOST, insecure=True)\n",
    "client.apply(asynchronous=False, verbose=True)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "01a8ffa0",
   "metadata": {},
   "source": [
    "## Follow the lineage stage by stage\n",
    "\n",
    "`client.dataframe()` computes and returns the DataFrame at any node in the graph. Reading them — the raw source, then each transformation derived from it — you can *see* every node that feeds the served feature. This is the lineage, made concrete."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "id": "7202f13e",
   "metadata": {
    "execution": {
     "iopub.execute_input": "2026-07-23T23:10:58.241785Z",
     "iopub.status.busy": "2026-07-23T23:10:58.241210Z",
     "iopub.status.idle": "2026-07-23T23:10:58.364238Z",
     "shell.execute_reply": "2026-07-23T23:10:58.363928Z"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "1. Raw source:\n",
      "No resources to apply\n"
     ]
    },
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>TransactionID</th>\n",
       "      <th>CustomerID</th>\n",
       "      <th>TransactionAmount</th>\n",
       "      <th>IsFraud</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>T00067</td>\n",
       "      <td>C1000</td>\n",
       "      <td>211.20</td>\n",
       "      <td>True</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>T00093</td>\n",
       "      <td>C1000</td>\n",
       "      <td>110.80</td>\n",
       "      <td>False</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>T00295</td>\n",
       "      <td>C1000</td>\n",
       "      <td>80.36</td>\n",
       "      <td>False</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>T00485</td>\n",
       "      <td>C1000</td>\n",
       "      <td>103.49</td>\n",
       "      <td>False</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>T00062</td>\n",
       "      <td>C1001</td>\n",
       "      <td>26.65</td>\n",
       "      <td>False</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "  TransactionID CustomerID  TransactionAmount  IsFraud\n",
       "0        T00067      C1000             211.20     True\n",
       "1        T00093      C1000             110.80    False\n",
       "2        T00295      C1000              80.36    False\n",
       "3        T00485      C1000             103.49    False\n",
       "4        T00062      C1001              26.65    False"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "2. After clean_transactions (invalid amounts gone):\n",
      "No resources to apply\n"
     ]
    },
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>CustomerID</th>\n",
       "      <th>TransactionAmount</th>\n",
       "      <th>IsFraud</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>C1000</td>\n",
       "      <td>211.20</td>\n",
       "      <td>True</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>C1000</td>\n",
       "      <td>110.80</td>\n",
       "      <td>False</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>C1000</td>\n",
       "      <td>80.36</td>\n",
       "      <td>False</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>C1000</td>\n",
       "      <td>103.49</td>\n",
       "      <td>False</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>C1001</td>\n",
       "      <td>26.65</td>\n",
       "      <td>False</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "  CustomerID  TransactionAmount  IsFraud\n",
       "0      C1000             211.20     True\n",
       "1      C1000             110.80    False\n",
       "2      C1000              80.36    False\n",
       "3      C1000             103.49    False\n",
       "4      C1001              26.65    False"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "3. After avg_user_transaction (aggregated per user):\n",
      "No resources to apply\n"
     ]
    },
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>user_id</th>\n",
       "      <th>avg_transaction_amt</th>\n",
       "      <th>transaction_count</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>C1047</td>\n",
       "      <td>118.431111</td>\n",
       "      <td>9</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>C1032</td>\n",
       "      <td>173.796667</td>\n",
       "      <td>6</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>C1006</td>\n",
       "      <td>118.632857</td>\n",
       "      <td>7</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>C1017</td>\n",
       "      <td>85.532500</td>\n",
       "      <td>8</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>C1023</td>\n",
       "      <td>116.270769</td>\n",
       "      <td>13</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "  user_id  avg_transaction_amt  transaction_count\n",
       "0   C1047           118.431111                  9\n",
       "1   C1032           173.796667                  6\n",
       "2   C1006           118.632857                  7\n",
       "3   C1017            85.532500                  8\n",
       "4   C1023           116.270769                 13"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "# NBVAL_SKIP\n",
    "print(\"1. Raw source:\")\n",
    "display(client.dataframe(transactions).head())\n",
    "\n",
    "print(\"2. After clean_transactions (invalid amounts gone):\")\n",
    "display(client.dataframe(clean_transactions).head())\n",
    "\n",
    "print(\"3. After avg_user_transaction (aggregated per user):\")\n",
    "display(client.dataframe(avg_user_transaction).head())"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "bff79b20",
   "metadata": {},
   "source": [
    "### See the lineage graph visually\n",
    "\n",
    "The Featureform dashboard at **http://localhost** renders the dependency DAG for every resource. Open the `avg_transactions` feature and you'll see its lineage `transactions → avg_user_transaction → avg_transactions`; the `clean_transactions` transformation (which feeds the label) hangs off the same `transactions` source — the sibling structure you just walked in code, plus variants, owners, and timestamps for audit."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "d5c3f0e7",
   "metadata": {},
   "source": [
    "## Serve the finished feature from Redis\n",
    "\n",
    "The end of the pipeline: request the feature for a single entity key. This read is served from Redis at low latency — the value having flowed through the whole traceable pipeline to get here. We grab a `user_id` that appears in the aggregated output above."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "id": "4bf5457c",
   "metadata": {
    "execution": {
     "iopub.execute_input": "2026-07-23T23:10:58.365728Z",
     "iopub.status.busy": "2026-07-23T23:10:58.365612Z",
     "iopub.status.idle": "2026-07-23T23:10:58.407671Z",
     "shell.execute_reply": "2026-07-23T23:10:58.406972Z"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "No resources to apply\n",
      "avg_transactions for user C1047: [118.43111419677734]\n"
     ]
    }
   ],
   "source": [
    "# NBVAL_SKIP\n",
    "user_id = client.dataframe(avg_user_transaction)[\"user_id\"].iloc[0]\n",
    "avg_txn = client.features(\n",
    "    [(\"avg_transactions\", \"quickstart\")],\n",
    "    {\"user\": user_id},\n",
    ")\n",
    "print(f\"avg_transactions for user {user_id}:\", avg_txn)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "e5771556",
   "metadata": {},
   "source": [
    "## Build a training set from the same definitions\n",
    "\n",
    "The offline side reuses the exact feature/label definitions. The dataset is iterable and streams rows of `(features, label)`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "id": "6731c466",
   "metadata": {
    "execution": {
     "iopub.execute_input": "2026-07-23T23:10:58.413411Z",
     "iopub.status.busy": "2026-07-23T23:10:58.413042Z",
     "iopub.status.idle": "2026-07-23T23:10:58.432281Z",
     "shell.execute_reply": "2026-07-23T23:10:58.431349Z"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[[126.4625]] -> [ True]\n",
      "[[126.4625]] -> [False]\n",
      "[[126.4625]] -> [False]\n",
      "[[126.4625]] -> [False]\n",
      "[[66.58714286]] -> [False]\n"
     ]
    }
   ],
   "source": [
    "# NBVAL_SKIP\n",
    "dataset = client.training_set(\"fraud_training\", \"quickstart\")\n",
    "\n",
    "for i, row in enumerate(dataset):\n",
    "    print(row.features(), \"->\", row.label())\n",
    "    if i >= 4:\n",
    "        break"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "577fcc70",
   "metadata": {},
   "source": [
    "## Cleanup\n",
    "\n",
    "Stop and remove the containers when you're done."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "id": "6f0c624e",
   "metadata": {
    "execution": {
     "iopub.execute_input": "2026-07-23T23:10:58.434663Z",
     "iopub.status.busy": "2026-07-23T23:10:58.434536Z",
     "iopub.status.idle": "2026-07-23T23:11:11.696149Z",
     "shell.execute_reply": "2026-07-23T23:11:11.694045Z"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Tearing down Featureform on Docker\n",
      "Stopping containers...\n",
      "\tStopping featureform container\n",
      "Container quickstart-clickhouse not found. Skipping...\n",
      "ff-clickhouse\n",
      "ff-redis\n",
      "featureform\n",
      "ff-net\n"
     ]
    }
   ],
   "source": [
    "# NBVAL_SKIP\n",
    "import sys\n",
    "# Tear down everything this notebook started. Remove containers (including the coordinator)\n",
    "# before the network; the coordinator's endpoint releases asynchronously, so retry the delete.\n",
    "!{sys.executable} -m featureform stop docker\n",
    "!docker rm -f ff-clickhouse ff-redis featureform 2>/dev/null\n",
    "!for i in $(seq 5); do docker network rm ff-net 2>/dev/null && break || sleep 1; done"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "f73d3f73",
   "metadata": {},
   "source": [
    "## Learn more\n",
    "\n",
    "- [Featureform transformations](https://docs.featureform.com/) — SQL and DataFrame transformations, chaining, and variants\n",
    "- [Featureform + Redis fraud detection recipe](./02_featureform_fraud_detection.ipynb) — a single-transformation version of this pattern"
   ]
  }
 ],
 "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
}