Production feature engineering patterns for ML pipelines. Covers feature stores, temporal features, automated feature selection, and data leakage prevention.
Feature engineering is the process of transforming raw data into structured inputs that machine learning models can interpret effectively. It matters when models fail to generalize beyond training data, when inference latency exceeds service SLOs, or when pipeline runs take hours to complete due to redundant or poorly structured feature computations.
Features must be consistently defined across training and inference. The schema governs how a feature is computed, stored, and consumed.
A feature is described by a feature.json file in the pipeline’s features/ directory:
{
"name": "user_avg_session_duration",
"type": "float",
"source": "user_sessions",
"computation": {
"method": "windowed_aggregate",
"window": "7d",
"aggregation": "mean",
"group_by": ["user_id"],
"timestamp_column": "start_time"
},
"materialization": {
"table": "feature_store.user_avg_session_duration",
"schedule": "0 0 * * *",
"format": "parquet",
"partitioning": ["date"]
},
"dependencies": ["user_sessions", "users"]
}The computation block defines how the feature is derived, while materialization specifies where and when it is computed. The dependencies list ensures the pipeline rebuilds the feature when any source changes.
Batch feature computation runs daily, populating a feature store. Real-time feature serving enables low-latency inference.
Batch computation uses dask for parallel processing:
dask-worker --scheduler-address tcp://scheduler:8786 --n-workers 8 --memory-limit 20GBA feature_pipeline.py script orchestrates the batch job:
import dask.dataframe as dd
from dask.distributed import Client
client = Client("tcp://scheduler:8786")
raw_sessions = dd.read_csv("s3://raw-data/sessions/*.csv")
# Compute feature
feature = (
raw_sessions
.groupby("user_id")
.agg({"session_duration": "mean"})
.rename(columns={"session_duration": "user_avg_session_duration"})
.reset_index()
)
# Write to feature store
feature.to_parquet(
"s3://feature-store/user_avg_session_duration/",
partition_on=["date"],
write_index=True
)
client.gather(client.compute(feature))For real-time feature serving, the pipeline uses a FeatureServer that exposes a gRPC endpoint:
service FeatureServer {
rpc GetFeatures(GetFeaturesRequest) returns (GetFeaturesResponse);
}
message GetFeaturesRequest {
repeated string feature_names = 1;
map<string, string> entity_ids = 2;
string timestamp = 3;
}
message GetFeaturesResponse {
map<string, FeatureValue> features = 1;
}
message FeatureValue {
oneof value {
float float_value = 1;
string string_value = 2;
int64 int_value = 3;
}
}A model inference request includes entity IDs and a timestamp. The server returns all requested features in a single round-trip, reducing latency.
Feature pipelines are governed by a Directed Acyclic Graph (DAG) that defines the order and inputs of computation.
A dag.yaml file defines the pipeline structure:
name: user_engagement_pipeline
tasks:
- name: compute_user_session_stats
inputs:
- source: user_sessions
version: 2023-09-15
outputs:
- table: feature_store.user_avg_session_duration
schema: features/user_avg_session_duration.json
depends_on: []
command: |
python feature_pipeline.py --source user_sessions --output feature_store/user_avg_session_duration
schedule: "0 0 * * *"
- name: compute_user_engagement_score
inputs:
- table: feature_store.user_avg_session_duration
version: latest
- table: feature_store.user_daily_active_count
version: 2023-09-15
outputs:
- table: feature_store.user_engagement_score
schema: features/user_engagement_score.json
depends_on:
- compute_user_session_stats
command: |
python calculate_engagement.py --session-stats feature_store/user_avg_session_duration --daily-active feature_store/user_daily_active_count
schedule: "0 1 * * *"The depends_on list ensures that a task runs only after all its dependencies have completed. The schedule field is a cron expression, and command is executed in a Docker container with the specified inputs mounted.
Each feature table is versioned via a version column, and each pipeline run records its run_id in a pipeline_runs table.
CREATE TABLE pipeline_runs (
run_id UUID PRIMARY KEY,
pipeline_name VARCHAR(255),
start_time TIMESTAMPTZ,
end_time TIMESTAMPTZ,
status VARCHAR(32),
inputs JSONB,
outputs JSONB,
error TEXT
);When a feature is materialized, the run_id is written to the feature table’s metadata:
INSERT INTO feature_store.user_avg_session_duration (
user_id,
user_avg_session_duration,
version,
run_id
)
VALUES (123, 142.3, '2023-09-15', 'a1b2c3d4-e5f6-7890-g1h2-i3j4k5l6m7n8');The version column tracks the feature schema and data used to compute it. Queries to the feature store include a version parameter to ensure consistency between training and inference.
A feature drift detection job runs weekly, comparing the current feature distribution to the baseline.
python drift_detector.py \
--feature_name user_avg_session_duration \
--current_data s3://feature-store/user_avg_session_duration/ \
--baseline_data s3://feature-store/user_avg_session_duration/2023-09-01/ \
--output s3://drift-reports/user_avg_session_duration/2023-09-15/The script computes statistical summaries (mean, median, standard deviation) and Kullback-Leibler divergence between the current and baseline distributions.
The output includes a drift_report.json:
{
"feature": "user_avg_session_duration",
"baseline": {
"mean": 110.2,
"std": 35.1,
"n": 450000
},
"current": {
"mean": 128.7,
"std": 41.3,
"n": 510000
},
"drift_score": 0.48,
"status": "drift_detected",
"threshold": 0.3
}A drift score above the threshold triggers an alert. The status is set to drift_detected, no_drift, or regression.
Missing features during inference cause model predictions to fail silently. A FeatureValidator checks input data before model inference.
from typing import Dict, List, Optional
def validate_features(
features: Dict[str, List[Optional[float]]],
required_features: List[str],
feature_schema: Dict[str, str]
) -> Dict[str, List[str]]:
errors = []
for feature_name, feature_values in features.items():
if feature_name not in required_features:
continue
expected_type = feature_schema[feature_name]
for i, value in enumerate(feature_values):
if value is None:
errors.append(f"{feature_name}[{i}]: missing value")
elif isinstance(value, float) and not (expected_type == "float"):
errors.append(f"{feature_name}[{i}]: expected {expected_type}, got float")
elif isinstance(value, int) and expected_type == "float":
errors.append(f"{feature_name}[{i}]: expected float, got int")
return {"errors": errors, "total": len(errors)}The validator is called during model serving:
response = requests.post(
"http://model-server:8000/predict",
json={
"user_ids": [1, 2, 3],
"features": {
"user_avg_session_duration": [142.3, None, 155.1],
"user_daily_active_count": [7, 5, 8]
}
}
)
if response.status_code == 400:
validation_error = response.json()
# Log error, retry, or serve fallback modelThe model server returns a 400 Bad Request with a validation error payload when required features are missing or invalid.
Feature computation is expensive. Materializing features in a columnar store reduces latency.
Each feature table is partitioned by date and stored as Parquet:
# Example: query a feature table
psql -c "
SELECT user_id, user_avg_session_duration
FROM feature_store.user_avg_session_duration
WHERE date >= '2023-09-01'
AND date < '2023-09-15'
AND user_id IN (1, 2, 3);
"The query uses partition pruning: only files for relevant dates are read. The table is columnar, so only the user_avg_session_duration column is loaded.
To reduce inference latency, the feature store caches frequently accessed features in Redis:
redis-cli -n 0
> HSET user_features:123 user_avg_session_duration 142.3 user_daily_active_count 7
> GET user_features:123The model server checks the cache before querying the feature store.
When a feature schema changes, old data must be re-computed to maintain consistency.
A backfill job re-computes features for a historical range:
python backfiller.py \
--feature user_avg_session_duration \
--start_date 2023-01-01 \
--end_date 2023-08-31 \
--parallelism 16 \
--output_format parquet \
--partitioning dateThe backfiller reads raw data for the given period and computes the feature using the new schema. It writes results to a new table, feature_store.user_avg_session_duration_v2, and updates the feature_schema table:
UPDATE feature_schema
SET version = 'v2',
last_updated = NOW(),
description = 'Updated computation to include session end time'
WHERE feature_name = 'user_avg_session_duration';After backfilling, a feature_migration job updates all references to the feature:
UPDATE model_configs
SET feature_sources = '{"user_avg_session_duration": "feature_store.user_avg_session_duration_v2"}'
WHERE model_name = 'user_engagement_model';This ensures that both training and inference use the new feature version.
user_avg_session_duration, but the feature store returns user_avg_session_duration_ms. The schema mismatch is silent until the model’s output is off.timestamp column used for windowed aggregation is in UTC, but the pipeline assumes local time. Aggregations include data from the wrong day.feature_schema table is not updated, so inference uses old feature definitions.user_id as a string, while the model uses user_id as an integer. The cache is effectively unused.feature_schema file. The report shows no drift, but the feature was not properly defined.FeatureValidator returns a list of errors, but the model server does not log them. The error is discovered only when users report incorrect predictions.These failures are caught by a feature_health_check cron job that runs every hour:
0 * * * * python health_check.py --check features --check drift --check cache --check schemaThe health check logs results to logs/health/feature_health_2023-09-15.log, which is monitored by Prometheus and Alertmanager.
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