Keeping training and serving honest: a feature store walkthrough with Feast, Flink and Kafka

Most ML outages I have seen were not model problems. The model was fine. What broke was the plumbing around it: a feature computed one way in the training notebook and a slightly different way in the service, a join that quietly leaked tomorrow's data into yesterday's training rows, a schema that drifted between the batch job and the API. The model learned something true about a world that does not exist at inference time.

feast-demo is a small, complete project I built to make that failure mode concrete and to show what a feature store does about it. It trains and serves a model that answers one question: will user user_id skip track track_id right now? Everything you need is in the repo — synthetic data, a Kafka cluster, a Flink job, batch jobs, training, an inference server — and it all runs on a local kind cluster with Nix and direnv. This post walks through it end to end.

The shape of the problem

The model is an XGBoost classifier that predicts skip probability from eleven features:

Entity	Feature	Source
user	`user_skip_rate_7d`, `user_plays_7d`, `user_avg_play_ms`	daily batch, 7-day sliding window
track	`track_lifetime_plays`, `track_lifetime_skip_rate`, `genre`, `tempo`	daily batch, lifetime
track	`track_plays_last_1h`, `track_skips_last_1h`	Flink, pushed every minute
request	`hour_of_day`, `is_weekend`	computed at request time

Three of those rows come from three different systems with three different latencies, and the same eleven numbers have to mean the same thing when we build the training set and when a request lands at /predict. That is the whole job of the feature store.

Why Feast

Feast solves two problems that look unrelated until you have been burned by both:

Feature consistency. The feature schema is declared once, in Python, as Entity, FeatureView, PushSource and FeatureService objects. That declaration is the contract every producer and consumer validates against: batch jobs, the stream job, training and the inference server.
Point-in-time correctness. When you assemble training rows, Feast joins each labelled event to the feature values that were true at the time of the event, not the latest ones. Get this wrong and the model sees the future during training, posts a wonderful AUC, and performs like a coin flip in production.

Here is the heart of the schema, from python/feast/features.py:

user = Entity(name="user", join_keys=["user_id"])
track = Entity(name="track", join_keys=["track_id"])

user_stats = FeatureView(
    name="user_stats",
    entities=[user],
    ttl=timedelta(days=2),
    source=PostgreSQLSource(name="user_stats_source",
                            query="SELECT * FROM user_stats",
                            timestamp_field="event_timestamp"),
    schema=[
        Field(name="user_skip_rate_7d", dtype=Float32),
        Field(name="user_plays_7d", dtype=Int64),
        Field(name="user_avg_play_ms", dtype=Float32),
    ],
)

# Pushed from the stream processor; the Postgres table is the offline fallback.
track_real_time_push = PushSource(
    name="track_realtime_push",
    batch_source=track_real_time_batch,
)

track_realtime = FeatureView(
    name="track_realtime",
    entities=[track],
    ttl=timedelta(hours=2),
    source=track_real_time_push,
    schema=[
        Field(name="track_plays_last_1h", dtype=Int64),
        Field(name="track_skips_last_1h", dtype=Int64),
    ],
)

@on_demand_feature_view(
    sources=[RequestSource(name="req", schema=[Field(name="request_ts_epoch", dtype=Int64)])],
    schema=[Field(name="hour_of_day", dtype=Int32), Field(name="is_weekend", dtype=Bool)],
)
def request_features(inp):
    ts = pd.to_datetime(inp["request_ts_epoch"], unit="s", utc=True)
    out = pd.DataFrame()
    out["hour_of_day"] = ts.dt.hour
    out["is_weekend"] = ts.dt.dayofweek >= 5
    return out

skip_svc = FeatureService(
    name="skip_v1",
    features=[user_stats, track_stats, track_realtime, request_features],
)

The FeatureService is the unit you train against and serve from. Versioning it (skip_v1, skip_v2, …) is how you change the feature set without breaking the model that is already in production.

Three stores back this up, configured in feature_store.yaml:

registry (Postgres) — where the schema metadata lives after feast apply.
offline store (Postgres) — historical feature rows, read when materializing and when building training sets.
online store (Redis) — the low-latency copy the inference server reads. Written by feast materialize and by pushes from the stream job.

Getting data in: batch, stream, and the lake

seed_history.py ──▶ S3 (raw/plays.parquet)
                        │
            ┌───────────┴────────────┐
            ▼                        ▼
       java/batch               java/producer ──▶ Kafka `plays`
  (DuckDB → Postgres)                 │                │
            │                         ▼                ▼
  feast materialize            java/stream       Kafka Connect
  (Postgres → Redis)          (Flink → push)     (→ S3 parquet)
            │                         │
            └──────────┬──────────────┘
                       ▼
                 java/serving
        (Feast online → ONNX → P(skip))

Seeding history

python/training/seed_history.py fabricates two million play events for 20 users and 500 tracks over 60 days, with a per-track skip prior so there is actually something to learn, and writes them as Parquet to the raw bucket. Track and play entities are deliberately not modelled as tables: everything downstream is derived from the event log.

Batch jobs: DuckDB over Parquet, into the offline store

java/batch turns raw events into the three offline tables. Each processor runs a window query with DuckDB directly against the Parquet in S3 and bulk loads the result into Postgres. The user stats job, for example, produces a daily snapshot of a 7-day sliding window:

WITH windows AS (
  SELECT unnest(generate_series(start_dt, end_dt, interval '1 day')) AS window_end
)
SELECT
  r.user_id,
  CAST(count(*) AS BIGINT)                                    AS user_plays_7d,
  CAST(avg(CASE WHEN r.skipped THEN 1.0 ELSE 0.0 END) AS FLOAT) AS user_skip_rate_7d,
  CAST(avg(r.played_ms) AS FLOAT)                             AS user_avg_play_ms,
  w.window_end                                                AS event_timestamp
FROM raw_events r
CROSS JOIN windows w
WHERE r.event_timestamp >= w.window_end - interval '7 days'
  AND r.event_timestamp <  w.window_end
GROUP BY r.user_id, w.window_end

Notice event_timestamp is the window end. That is what makes the point-in-time join work later: a training row from June 9th will be joined to the snapshot whose window closed on or before June 9th, never after.

The part I care most about is what happens before the query runs. Every batch job boots a FeastRegistry client that pulls the skip_v1 feature service from feast serve_registry and refuses to write a table whose columns do not match the declared view:

public void validateColumns(String viewName, Set<String> outputColumns) {
  FeatureViewSchema schema = getView(viewName);
  Set<String> expected = schema.expectedColumns(); // entity keys + features + timestamp
  if (!expected.equals(outputColumns)) {
    throw new RuntimeException(
        "Schema mismatch for feature view '%s':\n  Expected: %s\n  Actual: %s"
            .formatted(viewName, expected, outputColumns));
  }
}

The registry is the single source of truth, and a job that disagrees with it fails loudly before it can poison the offline store. The same client also checks that a processor exists for every view the feature service requires, so adding a view to skip_v1 without a job to fill it is a startup error, not a mystery NULL three weeks later.

Streaming: Flink, sliding windows, and the push source

track_plays_last_1h cannot wait for a nightly job. java/stream is a Flink job that consumes the plays topic, keys by track, aggregates a one-hour window that slides every minute, and pushes each result into Feast:

plays
    .keyBy(Play::trackId)
    .window(SlidingEventTimeWindows.of(Duration.ofHours(1), Duration.ofMinutes(1)))
    .aggregate(new CountAgg(), new EmitWindow())
    .map(new FeastPushFunction(SharedConfig.FEAST_PUSH));

Two details are worth pausing on.

Watermarks. The source uses WatermarkStrategy.forBoundedOutOfOrderness(Duration.ofSeconds(30)) with the event's own event_timestamp. A window for 11:59–12:00 does not emit the instant the clock hits 12:00; it waits for the watermark, so an event stamped 11:59:59 that arrives at 12:00:25 still lands in the right bucket. This is event-time processing for live data. Backfilling a dead mobile client's queue is a different job (a bounded batch job), and mixing the two is how you get "why is last hour's count wrong" tickets.

keyBy. Keying by trackId is what lets Flink spread the aggregate-and-push work across task managers safely: every track's materialised view is owned by exactly one key, so horizontal scaling does not change the answer.

The push itself is a plain HTTP POST to feast serve's push endpoint, naming the PushSource declared in the schema:

body.put("push_source_name", "track_realtime_push");
df.set("track_id", arrayOf(value.trackId));
df.set("track_plays_last_1h", arrayOf(value.plays));
df.set("track_skips_last_1h", arrayOf(value.skips));
df.set("event_timestamp", arrayOf(value.windowEnd));

Feast validates the payload against the view's schema and writes it to Redis. The stream job never talks to the online store directly, which is the point: the schema contract is enforced on the write path too. (A small Flink footnote: FeastPushFunction is serialised and shipped to workers, so the ObjectMapper and HttpClient are transient and created in open().)

Kafka Connect closes the loop

New events also need to end up in the lake, or tomorrow's batch jobs will only ever see the seeded history. A Kafka Connect S3 sink (k8s/kafka-connect-s3-sink.yaml) writes everything on plays into the raw bucket as partitioned Parquet. Batch recomputes from the lake, feast materialize pushes the result to Redis, and the online store converges with what the stream job already pushed. Same rows, two paths, one schema.

Training without leaking the future

python/training/train.py is short because Feast is doing the hard part. Labels come straight from the raw events — the feature store does not store labels:

labels = pd.read_parquet(f"s3://{config.bucket}/raw/plays.parquet", storage_options=MINIO)[
    ["user_id", "track_id", "event_timestamp", "skipped"]
].rename(columns={"skipped": "label"})
labels["request_ts_epoch"] = labels["event_timestamp"].astype("int64") // 10**9

Then one call builds the wide training frame:

store = FeatureStore(repo_path=".../feast")
training_df = store.get_historical_features(
    entity_df=labels,
    features=store.get_feature_service("skip_v1"),
).to_df()

For each labelled row, get_historical_features looks up user_stats, track_stats and track_realtime by their join keys and takes the newest feature row with timestamp <= event_timestamp (and within the view's TTL). The on-demand view computes hour_of_day and is_weekend from the same timestamp. The result is eleven features that are exactly what the serving path would have produced had the request arrived at that moment — which is the definition of no leakage.

To be clear about where that guarantee lives: it is a property of dataset construction, not of the training loop. XGBoost does not know or care. If you build the frame with a naive LEFT JOIN on the latest stats table, you get leakage no matter how carefully you train.

The model is exported to ONNX so the JVM can run it without a Python runtime:

model = xgb.XGBClassifier(n_estimators=200, max_depth=6, eval_metric="logloss")
model.fit(X.values, y.values)
onnx_model = convert_xgboost(model, initial_types=[("input", FloatTensorType([None, X.shape[1]]))])

Serving: the same eleven numbers

java/serving is a Spring Boot service with one endpoint. The controller fetches online features for the request's entities and a request-time timestamp, assembles the vector, and runs ONNX Runtime:

@PostMapping("/predict")
public PredictResponse predict(@RequestBody PredictRequest r) {
  long nowEpoch = Instant.now().getEpochSecond();
  Map<String, Object> online = feast.getOnlineFeatures(
      "skip_v1",
      Map.of("user_id", r.userId(), "track_id", r.trackId(), "request_ts_epoch", nowEpoch));
  float[] vec = model.assemble(online);
  double pSkip = model.predictProba(vec);
  log.emit(r.requestId(), r.userId(), r.trackId(), nowEpoch, online, pSkip);
  return new PredictResponse(r.requestId(), pSkip);
}

FeastReader calls feast serve's /get-online-features with the feature service name — the same skip_v1 that training used — so the server cannot accidentally ask for a different feature set. OnnxModel.assemble pins the column order to the order Feast returns for the service (user_stats, track_stats, track_realtime, request_features), which is the one remaining place training and serving are coupled by convention rather than by schema. I would make that explicit next: derive the order from the registry at startup and fail if it changes.

ServedFeatureLogger is the other half of the honesty story. It logs the exact feature vector each prediction saw. Join that log back to what get_historical_features reconstructs for the same (user, track, timestamp) and you have a direct measurement of training/serving skew: if the two disagree, something in the write path is not matching the read path, and you find out from a dashboard instead of from a slow decay in the business metric.

curl -s -X POST http://localhost:8080/predict \
  -H "Content-Type: application/json" \
  -d '{"requestId":"test-1","userId":"u_0002","trackId":"t_0126"}'

What this buys you

Stepping back, the system has one schema and four readers of it:

the batch jobs validate their output columns against the registry before loading the offline store;
the stream job pushes through Feast, which validates against the push source's view;
training reads the feature service with a point-in-time join;
serving reads the same feature service from the online store.

Add a feature and you touch the declaration, the job that produces it, and nothing else — the registry tells everything downstream. Remove one and the batch job fails at startup. Rename one and the stream push is rejected. None of that needs a code reviewer to remember that a column exists in three places.

Things I would do next, roughly in order: derive the serving feature order from the registry; add a bounded Flink job for backfills so late data has a proper home; turn the served-feature log into an automated skew check; and move the compute layer onto Ray so batch, training and stream share one scheduler (there is a design doc for that in the repo). Feast stays where it is in all of those — it is the contract, and the contract is the part that was worth building.