Near-Real-Time Recommendation Features with Debezium, Kafka and Databricks

Last updated: October 2026

A recommendation system that refreshes every six hours misses the moment that matters most. When a subscriber watches three true crime documentaries in a row on a Saturday morning, the useful time to suggest a fourth is that morning, not that afternoon.

This article describes the feature pipeline we built for a streaming service with about 4.2 million active subscribers, covered in our media streaming case study. Its recommendation engine ran as a batch job every six hours, models were deployed by hand through Docker, and retraining took three days. Project details come from that engagement. The code is a simplified illustration written for this article, updated for the Databricks and MLflow features current in October 2026.

Fresh features beat a more complex model

Our first model iteration used a deep learning architecture, and it performed about the same as a simpler hybrid of collaborative filtering and content-based methods. What moved engagement was getting fresh signals into the model within minutes. The simpler model with real-time features beat the complex model with six-hour-old features every time, so the engineering effort went into the feature pipeline.

Three groups of features did most of the work:

• Recency-weighted genre preferences over a rolling seven days, with recent viewing weighted three times as heavily as viewing from five or more days ago.
• Watch-time patterns by time of day and day of week for each content type, so that a subscriber who watches comedies on weeknights and documentaries on weekend mornings sees different suggestions at different times.
• Skip patterns, such as consistently abandoning a show within its first ten minutes. Skip patterns turned out to be one of the strongest negative signals in the set, and adding them improved recommendation relevance by about 11% in A/B testing.

Syncing content metadata with Debezium

Content metadata lived in PostgreSQL, and the previous approach was a nightly pg_dump. Debezium captures every insert, update and delete as a Kafka event, so when a show is reclassified or new content is added, the features reflect it within minutes. Setting it up for 12 PostgreSQL tables took about two days, including schema configuration.

Debezium's PostgreSQL connector reads PostgreSQL's logical decoding output. The database needs wal_level = logical and enough replication slots and WAL senders for each connector. Debezium's plugin.name setting defaults to decoderbufs, so set it to pgoutput explicitly; pgoutput is built into PostgreSQL 10 and later and needs no extra library, and managed services such as Amazon RDS require it.

{
  "name": "content-metadata-cdc",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "database.hostname": "<host>",
    "database.port": "5432",
    "database.user": "debezium",
    "database.password": "<from your Kafka Connect config provider>",
    "database.dbname": "catalog",
    "topic.prefix": "catalog",
    "plugin.name": "pgoutput",
    "slot.name": "catalog_debezium",
    "publication.name": "catalog_publication",
    "publication.autocreate.mode": "filtered",
    "table.include.list": "public.titles,public.genres,public.title_genres",
    "heartbeat.interval.ms": "10000"
  }
}

The operational risk is disk space on the database server. A replication slot retains every WAL segment the connector has not yet confirmed, including while the connector is stopped, and PostgreSQL documents that slots prevent removal of required WAL even when nothing is connected to them. If a connector is down for a long time, WAL can fill the disk.

Two settings limit that. In PostgreSQL 13 and later, max_slot_wal_keep_size caps how much WAL a slot can hold, after which the slot is marked invalid; the default of -1 means unlimited. PostgreSQL 18 adds idle_replication_slot_timeout, which invalidates slots that stay inactive for too long. Some guides, including Debezium's documentation, suggest wal_keep_size for this, but PostgreSQL describes that setting as the WAL kept for standby servers, and it does not cap a slot. An invalidated slot means the connector has to take a new snapshot, so the cap should be large enough to cover a realistic outage, and slot lag should be monitored. Debezium's heartbeat setting helps when the captured tables are quiet but the server is busy, because it lets the slot advance.

-- postgresql.conf (PostgreSQL 13+)
-- max_slot_wal_keep_size = 50GB

-- How much WAL each slot is holding back.
SELECT slot_name,
       active,
       wal_status,
       pg_size_pretty(pg_wal_lsn_diff(pg_current_wal_lsn(), restart_lsn)) AS retained_wal
FROM pg_replication_slots;

Computing features with Structured Streaming

User interactions, every play, pause, skip, completion and search, already lived in Kafka topics. A Structured Streaming job reads them, aggregates them per user, genre and day, and writes the results to a Delta feature table. One design detail matters here. With a watermark, an aggregation in append mode only emits a window once the window has closed, which for a one-day window means the next day. Delta as a sink does not support update mode, so the usual pattern is foreachBatch with a MERGE, which upserts the running totals every micro-batch.

from pyspark.sql import functions as F

events = (spark.readStream.format("kafka")
    .option("kafka.bootstrap.servers", BOOTSTRAP_SERVERS)
    .option("subscribe", "user-interactions")
    .option("startingOffsets", "latest")
    .load()
    .select(F.from_json(F.col("value").cast("string"), EVENT_SCHEMA).alias("e"))
    .select("e.*"))

daily = (events
    .withWatermark("event_ts", "30 minutes")
    .groupBy(F.window("event_ts", "1 day"), "user_id", "genre")
    .agg(F.sum(F.when(F.col("event_type") == "play", F.col("watch_seconds"))).alias("watch_seconds"),
         F.count(F.when((F.col("event_type") == "skip") & (F.col("position_seconds") < 600), 1)).alias("early_skips"))
    .select(F.col("window.start").alias("day_start"), "user_id", "genre", "watch_seconds", "early_skips"))

def upsert(batch_df, batch_id):
    batch_df.createOrReplaceTempView("updates")
    batch_df.sparkSession.sql("""
        MERGE INTO ml.features.user_genre_daily t
        USING updates s
          ON t.user_id = s.user_id AND t.genre = s.genre AND t.day_start = s.day_start
        WHEN MATCHED THEN UPDATE SET *
        WHEN NOT MATCHED THEN INSERT *
    """)

(daily.writeStream
    .foreachBatch(upsert)
    .outputMode("update")
    .option("checkpointLocation", "/Volumes/ml/features/checkpoints/user_genre_daily")
    .trigger(processingTime="1 minute")
    .start())

Each streaming query needs its own checkpoint location, and changing the source, the stateful operations or the sink type generally means starting from a new one. The recency weighting is then applied when features are read, which keeps the stored aggregates simple. A linear weight that falls from 3.0 today to 1.0 at five days or more matches the weighting described above:

SELECT user_id,
       genre,
       SUM(watch_seconds * GREATEST(1.0, 3.0 - 0.4 * DATEDIFF(current_date(), day_start))) AS weighted_watch_seconds
FROM ml.features.user_genre_daily
WHERE day_start >= date_sub(current_date(), 6)
GROUP BY user_id, genre;

If minutes are not fresh enough, Databricks announced general availability of real-time mode for Structured Streaming on 19 March 2026. It has tight constraints: Databricks Runtime 16.4 LTS or later on classic compute, update output mode only, and no Delta source or sink and no foreachBatch, so it suits pipelines that write features back to Kafka or a serving store.

Feature tables and online serving in 2026

In the project, the features landed in Delta tables that served both model training and real-time inference. On Databricks today, any Delta table in Unity Catalog with a primary key constraint can serve as a feature table. Adding a time-series column to the key lets training sets use point-in-time lookups, so each training example sees only the feature values that existed when the event happened, which prevents leakage from the future.

CREATE TABLE ml.features.user_genre_daily (
    user_id        BIGINT NOT NULL,
    genre          STRING NOT NULL,
    day_start      TIMESTAMP NOT NULL,
    watch_seconds  BIGINT,
    early_skips    BIGINT,
    CONSTRAINT pk_user_genre_daily PRIMARY KEY (user_id, genre, day_start TIMESERIES)
);

# Python: build a training set that respects event time
from databricks.feature_engineering import FeatureEngineeringClient, FeatureLookup
fe = FeatureEngineeringClient()
lookups = [FeatureLookup(table_name="ml.features.user_genre_daily",
                         lookup_key=["user_id", "genre"],
                         timestamp_lookup_key="impression_ts")]

For low-latency serving, check which online option your workspace uses. Databricks' legacy online tables were deprecated and stopped being accessible after 15 January 2026. The current option is Online Feature Stores, powered by Databricks Lakebase, which publish feature tables in triggered, continuous or snapshot mode; the triggered and continuous modes need Change Data Feed enabled on the source table.

Registering, serving and A/B testing models

The old deployment process, exporting the model, building a Docker image, pushing it to ECR and updating a Kubernetes deployment, took a data scientist four to five hours and was error-prone. With MLflow for tracking and the registry, and Databricks Model Serving for hosting, deployment dropped to about 15 minutes. Models retrain daily, retraining fell from three days to four hours, and inference latency stays under 100 ms at the 99th percentile while serving more than 50 million predictions a day.

One part of that workflow has changed since. MLflow deprecated model stages such as Staging and Production in version 2.9.0, and models registered in Unity Catalog do not support stages at all. The current pattern is an alias, such as champion, that points at a specific model version and can be moved when a new version is approved.

from mlflow import MlflowClient
client = MlflowClient()   # MLflow 3 defaults to the databricks-uc registry

# Point the champion alias at version 7 once it passes evaluation.
client.set_registered_model_alias("ml.recs.hybrid_recommender", "champion", 7)

# Load whatever version the alias points to.
import mlflow
model = mlflow.pyfunc.load_model("models:/ml.recs.hybrid_recommender@champion")

In the project, A/B tests between model versions were configured outside the application code instead of being hard-coded into the app. On Databricks this is done on the serving endpoint, which can host several served entities and route a percentage of traffic to each.

{
  "served_entities": [
    {"name": "current",    "entity_name": "ml.recs.hybrid_recommender", "entity_version": "7",
     "workload_size": "Small", "scale_to_zero_enabled": false},
    {"name": "challenger", "entity_name": "ml.recs.hybrid_recommender", "entity_version": "8",
     "workload_size": "Small", "scale_to_zero_enabled": false}
  ],
  "traffic_config": {
    "routes": [
      {"served_model_name": "current",    "traffic_percentage": 90},
      {"served_model_name": "challenger", "traffic_percentage": 10}
    ]
  }
}

The business outcome was a 23% increase in average session time. Recommendation latency fell from six hours to under five minutes.