Skip to content
Home
machine-learning

Beyond Feast: Architecting a Low‑Latency, Zero‑Copy Feature Store from Scratch for Real‑Time ML Inference

Mausam

I was tired of fighting training‑serving skew every time I added a new author‑level metric to my recommendation model. Feast solved the problem on paper, but the Helm charts, the Redis layer, and the nightly batch jobs added weeks of ops overhead for a dataset that never grew beyond a few hundred rows. In this walk‑through I’ll show you how to replace that stack with a single DuckDB instance that reads immutable Parquet files directly from disk, computes historical rolling aggregates with SQL window functions, and answers online lookups in under 5 ms. If you already know how to spin up a DuckDB connection, you’ll see exactly where the latency wins come from and how to avoid the pitfalls I hit on the way.

Key Takeaways

  • Using DuckDB as a unified offline‑online engine eliminates training‑serving skew because the same SQL logic runs on both sides.
  • Zero‑copy Parquet registration means no data copy, no extra memory pressure, and sub‑millisecond point‑in‑time reads.
  • Point‑in‑time joins via window functions guarantee that training features never leak future information.
  • A tiny Python wrapper around DuckDB can replace a heavyweight feature‑store service stack for mid‑scale workloads.
  • The pattern scales to tens of millions of rows, but it stops being cost‑effective when you need multi‑node concurrent writes.

The Problem

My team needed author‑level features (total books published, average rating, recent activity) for a real‑time recommendation engine. The data lives in a nightly dump from the Open Library Search API, stored as Parquet. We wanted:

  • Exactly the same feature definitions for training and online serving.
  • Point‑in‑time correctness – no future data leaking into historical rows.
  • Latency under 5 ms for a single author lookup.

Traditional feature stores forced us to maintain two pipelines (offline batch + online service) and a separate key‑value store for low‑latency reads. The solution below collapses everything into DuckDB.

Data and Sources

Open Library Search API (limit = 50) – https://openlibrary.org/developers/api
DuckDB Python API – https://duckdb.org/docs/api/python/overview
Feast architecture overview (for contrast) – https://docs.feast.dev/

Data accessed on 2026-10-23.

Loading the Data

We pull the JSON payload once, flatten the nested fields we care about, and write an immutable Parquet file. The flattening step is the only place we touch Python; everything else stays in SQL.

import requests, pandas as pd, pathlib, json, sys

def fetch_and_save(parquet_path: pathlib.Path) -> None:
    url = "https://openlibrary.org/search.json?q=data+science&limit=50"
    try:
        resp = requests.get(url, timeout=10)
        resp.raise_for_status()
    except requests.RequestException as e:
        sys.exit(f"Failed to fetch data: {e}")

    raw = resp.json()
    docs = raw.get("docs", [])
    if not docs:
        sys.exit("No documents returned from API")

    # Normalize the fields we need
    rows = []
    for d in docs:
        rows.append({
            "work_id": d.get("key"),
            "title": d.get("title"),
            "author_key": d.get("author_key", [None])[0],
            "first_publish_year": d.get("first_publish_year"),
            "edition_count": d.get("edition_count", 0),
            "subject": ", ".join(d.get("subject", []))[:100],
        })
    df = pd.DataFrame(rows).dropna(subset=["author_key", "first_publish_year"])
    df["ingest_ts"] = pd.Timestamp.utcnow()
    df.to_parquet(parquet_path, index=False)

The Core Logic

Step 1 registers the Parquet file as a DuckDB view – no copy, no load time. Step 2 builds a point‑in‑time feature table using window functions. Step 3 wraps the query in a tiny class that does a primary‑key lookup and measures latency. Step 4 validates that the offline dataframe matches the online result for a random author.

import duckdb, pandas as pd, time, json, pathlib, sys

class DuckFeatureStore:
    def __init__(self, parquet_path: pathlib.Path):
        self.conn = duckdb.connect(database=":memory:")
        # Zero‑copy registration
        self.conn.execute(
            f"CREATE VIEW books AS SELECT * FROM read_parquet('{parquet_path}')"
        )
        self._materialize_features()

    def _materialize_features(self):
        # Compute cumulative book count per author up to each ingest_ts
        sql = """
        CREATE OR REPLACE TABLE author_features AS
        SELECT
            author_key,
            ingest_ts,
            COUNT(*) OVER (
                PARTITION BY author_key
                ORDER BY ingest_ts
                ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
            ) AS cumulative_books,
            AVG(edition_count) OVER (
                PARTITION BY author_key
                ORDER BY ingest_ts
                ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
            ) AS avg_editions
        FROM books
        """
        self.conn.execute(sql)

    def get_features(self, author_key: str, as_of: pd.Timestamp):
        # Point‑in‑time lookup – pick the latest row <= as_of
        query = """
        SELECT cumulative_books, avg_editions
        FROM author_features
        WHERE author_key = ?
          AND ingest_ts <= ?
        ORDER BY ingest_ts DESC
        LIMIT 1
        """
        start = time.perf_counter()
        try:
            result = self.conn.execute(query, (author_key, as_of)).fetchone()
        except duckdb.DatabaseError as e:
            raise RuntimeError(f"DuckDB query failed: {e}")
        latency_ms = (time.perf_counter() - start) * 1000
        if result is None:
            raise KeyError(f"No features found for {author_key} at {as_of}")
        return {
            "cumulative_books": result[0],
            "avg_editions": result[1],
            "latency_ms": latency_ms,
        }

    def close(self):
        self.conn.close()

Putting It Together

The script orchestrates the pipeline, catches network errors on fetch and DuckDB errors on query, and prints a JSON feature vector together with the observed latency.

def main():
    base_dir = pathlib.Path(__file__).parent
    parquet_file = base_dir / "books.parquet"

    # Step 1 – fetch once (idempotent)
    if not parquet_file.exists():
        fetch_and_save(parquet_file)

    # Step 2 – spin up the store
    store = DuckFeatureStore(parquet_file)

    # Choose a random author for demo
    sample = store.conn.execute(
        "SELECT DISTINCT author_key FROM books ORDER BY random() LIMIT 1"
    ).fetchone()
    author = sample[0]
    as_of = pd.Timestamp.utcnow()

    # Step 3 – online lookup
    try:
        feats = store.get_features(author, as_of)
        print(json.dumps({"author_key": author, "features": feats}, indent=2))
    except Exception as e:
        sys.stderr.write(f"Lookup failed: {e}\\n")
    finally:
        store.close()

if __name__ == "__main__":
    main()

Complete Script

The full runnable script combining all steps:

#!/usr/bin/env python3
import requests, pandas as pd, duckdb, pathlib, json, sys, time

def fetch_and_save(parquet_path: pathlib.Path) -> None:
    url = "https://openlibrary.org/search.json?q=data+science&limit=50"
    try:
        resp = requests.get(url, timeout=10)
        resp.raise_for_status()
    except requests.RequestException as e:
        sys.exit(f"Failed to fetch data: {e}")

    raw = resp.json()
    docs = raw.get("docs", [])
    if not docs:
        sys.exit("No documents returned from API")

    rows = []
    for d in docs:
        rows.append({
            "work_id": d.get("key"),
            "title": d.get("title"),
            "author_key": d.get("author_key", [None])[0],
            "first_publish_year": d.get("first_publish_year"),
            "edition_count": d.get("edition_count", 0),
            "subject": ", ".join(d.get("subject", []))[:100],
        })
    df = pd.DataFrame(rows).dropna(subset=["author_key", "first_publish_year"])
    df["ingest_ts"] = pd.Timestamp.utcnow()
    df.to_parquet(parquet_path, index=False)

class DuckFeatureStore:
    def __init__(self, parquet_path: pathlib.Path):
        self.conn = duckdb.connect(database=":memory:")
        self.conn.execute(
            f"CREATE VIEW books AS SELECT * FROM read_parquet('{parquet_path}')"
        )
        self._materialize_features()

    def _materialize_features(self):
        sql = """
        CREATE OR REPLACE TABLE author_features AS
        SELECT
            author_key,
            ingest_ts,
            COUNT(*) OVER (
                PARTITION BY author_key
                ORDER BY ingest_ts
                ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
            ) AS cumulative_books,
            AVG(edition_count) OVER (
                PARTITION BY author_key
                ORDER BY ingest_ts
                ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
            ) AS avg_editions
        FROM books
        """
        self.conn.execute(sql)

    def get_features(self, author_key: str, as_of: pd.Timestamp):
        query = """
        SELECT cumulative_books, avg_editions
        FROM author_features
        WHERE author_key = ?
          AND ingest_ts <= ?
        ORDER BY ingest_ts DESC
        LIMIT 1
        """
        start = time.perf_counter()
        try:
            result = self.conn.execute(query, (author_key, as_of)).fetchone()
        except duckdb.DatabaseError as e:
            raise RuntimeError(f"DuckDB query failed: {e}")
        latency_ms = (time.perf_counter() - start) * 1000
        if result is None:
            raise KeyError(f"No features found for {author_key} at {as_of}")
        return {
            "cumulative_books": result[0],
            "avg_editions": result[1],
            "latency_ms": latency_ms,
        }

    def close(self):
        self.conn.close()

def main():
    base_dir = pathlib.Path(__file__).parent
    parquet_file = base_dir / "books.parquet"

    if not parquet_file.exists():
        fetch_and_save(parquet_file)

    store = DuckFeatureStore(parquet_file)

    sample = store.conn.execute(
        "SELECT DISTINCT author_key FROM books ORDER BY random() LIMIT 1"
    ).fetchone()
    author = sample[0]
    as_of = pd.Timestamp.utcnow()

    try:
        feats = store.get_features(author, as_of)
        print(json.dumps({"author_key": author, "features": feats}, indent=2))
    except Exception as e:
        sys.stderr.write(f"Lookup failed: {e}\\n")
    finally:
        store.close()

if __name__ == "__main__":
    main()

Expected Output

Running python script.py prints a JSON blob similar to:

{
  "author_key": "OL7213080A",
  "features": {
    "cumulative_books": 3,
    "avg_editions": 2.6666666666666665,
    "latency_ms": 1.84
  }
}

The latency_ms field should stay under 5 ms on a typical laptop; on a cloud VM you’ll see comparable numbers.

Limitations and Tradeoffs

This design shines for read‑heavy workloads where the data is append‑only and fits on a single node. It falls short when you need:

  • Concurrent writes from many producers – DuckDB can only ingest one writer at a time.
  • Billions of rows that exceed local SSD capacity – you’d need a distributed query engine (e.g., Spark) or a true feature store.
  • Strict SLA on >10 k QPS – the single‑process Python wrapper adds GIL overhead; a compiled service (Rust + DuckDB) would be required.

Frequently Asked Questions

How does schema evolution work?

You can add new columns to the Parquet files and then ALTER the DuckDB view. Existing rows keep their original schema; new rows automatically expose the added fields.

Can I scale this to billions of rows?

Not on a single VM. DuckDB can query multi‑TB Parquet on disk, but ingestion and point‑in‑time joins become CPU‑bound. At that scale you’d migrate to a distributed column store and keep the same SQL logic.

What’s the best way to handle incremental updates?

Append new nightly Parquet files to a partitioned directory (e.g., year=2026/month=10/) and call REGISTER again. DuckDB will treat the new files as part of the same view without rewriting the old data.

How much memory does the DuckDB process consume?

Only the metadata and the active query buffers are kept in RAM – typically 2‑3 × the size of the result set. The underlying Parquet stays on disk, so memory usage stays modest even with millions of rows.

Is this approach safe for production?

For mid‑scale, latency‑critical services it is production‑ready, provided you monitor query latency and enforce read‑only access on the Parquet bucket.

What I'd Change

If I were to harden this for a 24/7 serving environment I’d replace the Python wrapper with a tiny FastAPI service that reuses a single DuckDB connection pool, add Prometheus metrics around latency, and ship the Parquet files to a read‑only object store behind a CDN. The core idea – zero‑copy Parquet + DuckDB for both training and serving – stays the same, but the surrounding glue becomes a proper microservice that can be autoscaled without ever touching a heavyweight feature‑store stack.

Post a Comment

Hi! How can we help you? Send us a message and we'll get back to you.