Production-grade guide to recommendation system architecture covering architecture patterns, implementation strategies, testing approaches, and operational best practices for enterprise engineering teams.
Recommendation systems are the backbone of user engagement in digital platforms, transforming raw interactions into personalized experiences. They matter when content, products, or services must be surfaced efficiently to individuals at scale—whether in e-commerce, media streaming, social networking, or search. A robust architecture must handle real-time scoring, batch processing, feedback loops, and model iteration, all while adapting to changing user behavior and content availability.
A recommendation system architecture is defined by the flow of data from raw signals to ranked outputs. The system must ingest user and item features, capture interactions, compute relevance scores, and serve results with low latency. The foundational layer consists of a data ingestion pipeline, feature store, model serving, and ranking engine.
User interactions—clicks, views, purchases, likes, time spent—are captured in event logs. These logs are ingested into a stream-processing engine using a reliable, scalable protocol like Apache Kafka or Amazon Kinesis. Each event is a JSON object with a consistent schema:
{
"event_id": "a1b2c3d4",
"user_id": 42017,
"item_id": 9876,
"timestamp": "2024-05-15T14:32:17Z",
"event_type": "click",
"session_id": "sess-88392"
}Events are published to Kafka topics such as user_events, item_events, and interaction_logs. Consumers process these events to build user profiles, update item metadata, and trigger model training.
A typical ingestion job runs in a containerized environment using Docker and orchestrated via Kubernetes. The consumer is a Python service using confluent-kafka:
from confluent_kafka import Consumer
import json
consumer = Consumer({
'bootstrap.servers': 'kafka1:9092,kafka2:9092',
'group.id': 'recommendation-ingest',
'auto.offset.reset': 'earliest'
})
consumer.subscribe(['user_events'])
while True:
msg = consumer.poll(timeout=1.0)
if msg is None:
continue
if msg.error():
print(f"Consumer error: {msg.error()}")
continue
event = json.loads(msg.value())
# Write to downstream systems: PostgreSQL, Redis, or a feature store
with psql.connect("dbname=recommendations user=app") as conn:
with conn.cursor() as cur:
cur.execute(
"INSERT INTO interactions (user_id, item_id, event_type, timestamp) "
"VALUES (%s, %s, %s, %s)",
(event['user_id'], event['item_id'], event['event_type'], event['timestamp'])
)
conn.commit()Failure modes: if the Kafka consumer fails to commit offsets, events are reprocessed; if the PostgreSQL connection drops, messages accumulate in a retry queue.
Features are the inputs to the recommendation model. They are stored in a centralized feature store accessible to training and serving. Features include user-level (e.g., total_views, avg_engagement_time), item-level (popularity_score, category_avg_rating), and context-specific (time_since_last_visit, device_type).
Features are computed and registered in batches or in real time. A feature pipeline uses a SQL-based engine (e.g., dbt, Spark SQL) to materialize features into tables.
-- Compute user feature: average time spent per session
CREATE TABLE user_features AS
SELECT
user_id,
AVG(time_spent) AS avg_time_per_session,
COUNT(*) AS total_sessions,
MAX(last_session_timestamp) AS last_activity
FROM user_sessions
GROUP BY user_id;Features are versioned and tagged. A feature is identified by a unique name and version:
user_features/avg_time_per_session/v3
item_features/popularity_score/v2A feature registry is maintained as a metadata table:
CREATE TABLE feature_registry (
feature_name TEXT PRIMARY KEY,
version VARCHAR(20),
source_table TEXT,
pipeline_query TEXT,
created_at TIMESTAMPTZ,
description TEXT
);The feature store is queried at serving time using a gRPC or REST API. A typical request:
GET /v1/feature/user_features/avg_time_per_session/v3?user_id=42017 HTTP/1.1
Host: feature-store.example.com
Authorization: Bearer <token>Response:
{
"feature": "user_features/avg_time_per_session/v3",
"value": 142.3,
"timestamp": "2024-05-15T14:30:00Z"
}Silent failure: if the feature version is not found, the system defaults to the latest version but logs a warning.
Recommendation models are trained on historical interaction data. The training pipeline is batch-oriented, using tools like Apache Spark or Dask, and runs daily or hourly.
A typical training job reads from a fact table of interactions and materialized features:
from pyspark.sql import SparkSession
spark = SparkSession.builder.appName("RecommendationTraining").getOrCreate()
interactions = spark.read.table("interactions")
user_features = spark.read.table("user_features")
item_features = spark.read.table("item_features")
training_data = interactions \
.join(user_features, on="user_id") \
.join(item_features, on="item_id") \
.withColumn("label", col("event_type").eq("purchase").cast("double")) \
.select("user_id", "item_id", "avg_time_per_session", "popularity_score", "label")
# Train model using LightGBM
from lightgbm import LGBMClassifier
from sklearn.model_selection import train_test_split
X_train, X_test, y_train, y_test = train_test_split(
training_data.drop("label").toPandas(),
training_data.select("label").toPandas(),
test_size=0.2
)
model = LGBMClassifier(n_estimators=100, learning_rate=0.1)
model.fit(X_train, y_train)
# Save model to disk
import joblib
joblib.dump(model, "/models/recommendation/v1.2.pkl")Model evaluation is done on holdout data. Metrics include AUC, precision@k, recall@k, and NDCG.
from sklearn.metrics import roc_auc_score, precision_score, ndcg_score
y_pred = model.predict_proba(X_test)[:, 1]
auc = roc_auc_score(y_test, y_pred)
# Precision at top 10 items
top_k = 10
precision_at_k = precision_score(y_test, (y_pred > 0.5).astype(int), k=top_k)
# NDCG@10
ndcg = ndcg_score(y_test.reshape(1, -1), y_pred.reshape(1, -1), k=top_k)Models are versioned and registered in a model registry. A model is identified by name and version:
recommendation_model/batch/lightgbm/v1.2The registry is a table:
CREATE TABLE model_registry (
model_name TEXT PRIMARY KEY,
version TEXT,
artifact_path TEXT,
created_at TIMESTAMPTZ,
metrics JSONB,
trained_on DATETIME,
description TEXT
);When a new model is trained, it is evaluated and promoted to production if it exceeds a threshold (e.g., AUC > 0.85).
The serving pipeline delivers recommendations to users in real time and batch. It consists of a ranking engine, a cache layer, and a serving API.
For real-time recommendations, a user requests a list of items. The system computes scores for candidate items and returns a ranked list.
A typical request:
POST /v1/recommendations HTTP/1.1
Host: recommendation-service.example.com
Content-Type: application/json
Authorization: Bearer <user-token>
{
"user_id": 42017,
"context": {
"device_type": "mobile",
"time_of_day": "evening",
"location": "San Francisco"
},
"num_results": 20
}The response:
{
"user_id": 42017,
"timestamp": "2024-05-15T14:33:10Z",
"recommendations": [
{
"item_id": 9876,
"score": 0.93,
"rank": 1,
"reason": "user_profile_match"
},
{
"item_id": 10123,
"score": 0.87,
"rank": 2,
"reason": "popular_item"
}
]
}The service uses a Python backend with FastAPI and a caching layer (Redis):
from fastapi import FastAPI
import redis
app = FastAPI()
r = redis.Redis(host="redis.example.com", port=6379, db=0)
@app.post("/v1/recommendations")
async def recommend(request: RecommendationRequest):
user_id = request.user_id
context = request.context
num_results = request.num_results
# Try cache first
cache_key = f"recs:{user_id}:{context['device_type']}:{num_results}"
cached = r.get(cache_key)
if cached:
return json.loads(cached)
# Fetch features
user_features = await fetch_feature("user_features/avg_time_per_session/v3", user_id)
item_features = await fetch_feature("item_features/popularity_score/v2", None)
# Load model from disk
model = joblib.load("/models/recommendation/v1.2.pkl")
# Generate candidate items
candidates = await get_candidate_items(user_id, context)
# Score candidates
features = []
for item in candidates:
row = {
"user_id": user_id,
"item_id": item["item_id"],
"avg_time_per_session": user_features["value"],
"popularity_score": item["popularity_score"]
}
features.append(row)
X = pd.DataFrame(features)
scores = model.predict_proba(X)[:, 1]
# Combine and sort
results = sorted(zip(candidates, scores), key=lambda x: x[1], reverse=True)[:num_results]
# Format response
response = {
"user_id": user_id,
"timestamp": datetime.now().isoformat(),
"recommendations": [
{
"item_id": item["item_id"],
"score": score,
"rank": i + 1,
"reason": "model_score"
}
for i, (item, score) in enumerate(results)
]
}
# Cache response
r.setex(cache_key, 3600, json.dumps(response))
return responseFailure modes: if the model file is missing, the service falls back to a default model. If Redis is unreachable, the system uses the cache as a secondary store.
Batch recommendations are precomputed and stored. They are used for email digests, homepage banners, or nightly updates.
A batch job runs nightly:
# Run in a Kubernetes job
kubectl run recommendation-batch-job --image=python:3.11 \
--command -- python3 /app/batch_recommendations.py \
--start-date "2024-05-14" \
--end-date "2024-05-15"The script:
def generate_batch_recommendations(start_date, end_date):
with psql.connect("dbname=recommendations user=app") as conn:
cur = conn.cursor()
cur.execute("""
SELECT user_id
FROM users
WHERE last_active > %s
ORDER BY total_engagement_score DESC
LIMIT 10000
""", (start_date,))
user_ids = [row[0] for row in cur.fetchall()]
# For each user, compute recommendations
results = []
for user_id in user_ids:
recs = recommend(user_id, {"device_type": "desktop", "time_of_day": "morning"})
results.append({
"user_id": user_id,
"recommendations": recs["recommendations"],
"generated_at": datetime.now()
})
# Save to table
with psql.connect("dbname=reports user=app") as conn:
with conn.cursor() as cur:
for res in results:
cur.execute("""
INSERT INTO user_recommendations_batch (user_id, recommendations, generated_at)
VALUES (%s, %s, %s)
""", (res["user_id"], res["recommendations"], res["generated_at"]))
conn.commit()Batch results are indexed and served via a REST API or stored in a data warehouse (e.g., Snowflake, BigQuery).
Recommendations are not static. The system must collect feedback and use it to improve models.
Feedback is captured as explicit and implicit signals:
These are ingested into the same event pipeline. A feedback loop runs daily:
-- Compute click-through rate for recommendations
CREATE TABLE recommendation_performance AS
SELECT
r.user_id,
r.item_id,
COUNT(*) AS impressions,
SUM(CASE WHEN e.event_type = 'click' THEN 1 ELSE 0 END) AS clicks,
AVG(e.time_spent) AS avg_time
FROM recommendations r
JOIN user_events e ON r.recommendation_id = e.recommendation_id
WHERE e.timestamp >= '2024-05-15'
GROUP BY r.user_id, r.item_id;The performance data is used to:
The system tracks model drift and data quality. A monitoring job checks for:
item_id in events)When a model fails to meet thresholds, it is flagged and rolled back.
FeatureMissingError: item_id not found in input.ScoreLatencyExceeded: 780ms.The architecture is not just a pipeline—it is a system that observes, adapts, and corrects itself.
This page was rewritten on 17 October 2026. It replaced a templated version that was not specific to its own title. How these pages are written and checked. If anything here is wrong, tell us at [email protected] and we will correct it.
We use cookies for analytics (Google Analytics) and advertising (Google AdSense) to improve your experience and support free content. Privacy Policy