When Batch Features Aren't Fast Enough
Most ML features are computed in batch. An Airflow job runs nightly, reads from the data warehouse, computes features, and writes them to a feature store. For many use cases, that's perfectly fine. Yesterday's features are close enough to today's reality.
But some models need features that are minutes or seconds old. Fraud detection needs the number of transactions in the last five minutes. Real-time pricing needs the current demand signal. Recommendation models need to react to what a user just clicked, not what they clicked yesterday.
Apache Flink is the tool I reach for when batch features aren't fast enough. It processes streaming data with exactly-once semantics, manages state across millions of keys, and integrates with the event sources (Kafka, Kinesis) and sinks (feature stores, databases) that ML infrastructure typically uses.
Streaming Feature Computation Patterns
The most common streaming features for ML fall into a few categories: windowed aggregations (count, sum, average over a time window), session-based features (metrics within a user session), and event sequence features (patterns in the order of events).
// Flink streaming features for fraud detection (Scala)
case class Transaction(
userId: String,
merchantId: String,
amount: Double,
timestamp: Long,
country: String
)
case class UserFeatures(
userId: String,
txnCountLast5Min: Long,
txnCountLast1Hour: Long,
avgAmountLast1Hour: Double,
distinctMerchantsLast1Hour: Long,
distinctCountriesLast5Min: Long,
maxAmountLast5Min: Double,
computedAt: Long
)
val transactions: DataStream[Transaction] = env
.addSource(new FlinkKafkaConsumer("transactions", schema, props))
.assignTimestampsAndWatermarks(
WatermarkStrategy
.forBoundedOutOfOrderness(Duration.ofSeconds(10))
.withTimestampAssigner((txn, _) => txn.timestamp)
)
// 5-minute sliding window features
val fiveMinFeatures = transactions
.keyBy(_.userId)
.window(SlidingEventTimeWindows.of(Time.minutes(5), Time.seconds(30)))
.aggregate(new TransactionAggregator())State Management for ML Features
Streaming features require state. The "number of transactions in the last 5 minutes" feature needs to remember the last 5 minutes of transactions for each user. For a service with millions of users, that state adds up quickly.
Flink manages state on RocksDB backends, which spills to disk when memory fills up. This is critical for ML features because the state size scales with the number of entities (users, merchants, devices) times the window size times the number of features. A naive implementation that keeps all events in memory will crash long before you reach production scale.
We covered a related topic in Model Registry Architecture and Versioning for Multi-Team Or.
class FraudFeatureFunction
extends KeyedProcessFunction[String, Transaction, UserFeatures] {
private var recentTxns: ListState[Transaction] = _
private var runningCount: ValueState[Long] = _
private var runningSum: ValueState[Double] = _
override def open(params: Configuration): Unit = {
val txnDesc = new ListStateDescriptor[Transaction](
"recent-txns", classOf[Transaction]
)
// TTL ensures state cleanup for inactive users
txnDesc.enableTimeToLive(
StateTtlConfig.newBuilder(Time.hours(2))
.setUpdateType(StateTtlConfig.UpdateType.OnReadAndWrite)
.build()
)
recentTxns = getRuntimeContext.getListState(txnDesc)
runningCount = getRuntimeContext.getState(
new ValueStateDescriptor("count", classOf[Long])
)
runningSum = getRuntimeContext.getState(
new ValueStateDescriptor("sum", classOf[Double])
)
}
override def processElement(
txn: Transaction,
ctx: Context,
out: Collector[UserFeatures]
): Unit = {
recentTxns.add(txn)
val prevCount = Option(runningCount.value()).getOrElse(0L)
runningCount.update(prevCount + 1)
val prevSum = Option(runningSum.value()).getOrElse(0.0)
runningSum.update(prevSum + txn.amount)
ctx.timerService().registerEventTimeTimer(
txn.timestamp + 300_000
)
val features = computeFeatures(txn.userId, txn.timestamp)
out.collect(features)
}
}The TTL configuration is essential. Without it, state for users who stop transacting stays in RocksDB indefinitely, consuming storage and slowing checkpoint times. A 2-hour TTL for fraud features means we keep state for active users and clean up inactive ones automatically.
Exactly-Once Delivery to the Feature Store
ML features need consistency. If a streaming job fails and restarts, features shouldn't be computed twice (double-counting) or skipped (missing data). Flink's checkpointing mechanism provides exactly-once state consistency within the job, but the sink connector to your feature store needs to cooperate.
For Redis-backed feature stores, idempotent writes (SET operations with the same key and value) give you effective exactly-once behavior. For databases, upsert operations achieve the same thing. The key is that re-processing the same event produces the same feature value and writes it to the same key, so duplicates are harmless.
Backfill and Feature Consistency
One of the trickiest problems with streaming features is backfill. When you add a new streaming feature, you need historical values for model training. But the streaming job only produces features going forward.
We solve this by maintaining a parallel batch computation path. Every streaming feature has a batch equivalent that computes the same value from historical data in the warehouse. The batch version runs on Spark and produces the same output schema as the streaming version. During training, we use batch features. During inference, we use streaming features. The feature store abstracts this — the model requests a feature by name and gets whichever version is freshest.
Related reading: Retrieval-Augmented Generation Architecture for Enterprise S.
Consistency between batch and streaming computations is validated weekly. We compare the streaming features against batch-recomputed values for the same time windows and flag any divergence above 0.1%. Divergence usually means a bug in one of the implementations, not a fundamental inconsistency in the approach.
Operational Considerations
Flink jobs need the same operational attention as production services. Monitoring covers processing latency (how far behind real-time the job is), checkpoint duration and size, and backpressure indicators. If the job falls behind, features become stale, and the model is effectively running on batch features again — defeating the whole purpose.
We set alerts on processing lag above 30 seconds and checkpoint failures. A lag alert means the job can't keep up with the incoming event rate and needs more parallelism or optimization. A checkpoint failure means the job might lose state on restart, which corrupts feature values.
Scaling Flink parallelism is straightforward for stateless operations but complicated for stateful ones. Increasing parallelism requires state redistribution across the new task instances, which means a job restart. We overprovision Flink parallelism at deploy time to handle expected traffic growth and only scale up during planned maintenance windows.
One more operational note: Flink's web UI is your best friend for diagnosing performance issues. The backpressure tab shows which operators are bottlenecks. The checkpoint history shows whether state size is growing faster than expected. And the task manager logs often contain the first clue when something goes wrong. Invest the time to understand that UI before you need it in an incident.
This connects to the ideas in Data Pipeline Orchestration for ML: Airflow, Prefect, and Da.
Feature Serving Architecture
Streaming features need to be available to the model at inference time with low latency. The standard pattern is to write computed features to a low-latency store (Redis, DynamoDB, or a purpose-built feature store) and have the serving application read from that store when building the feature vector for a prediction request.
# Writing streaming features to Redis feature store
from redis import Redis
class FeatureStoreSink:
def __init__(self, redis_host, redis_port, ttl_seconds=7200):
self.redis = Redis(host=redis_host, port=redis_port)
self.ttl = ttl_seconds
def write_features(self, user_id: str, features: dict):
key = f"features:user:{user_id}"
# Store as hash for individual field access
self.redis.hset(key, mapping={
k: str(v) for k, v in features.items()
})
self.redis.expire(key, self.ttl)
def read_features(self, user_id: str, fields: list) -> dict:
key = f"features:user:{user_id}"
values = self.redis.hmget(key, fields)
return {
field: float(val) if val isn't None else None
for field, val in zip(fields, values)
}The TTL on the feature store entries serves two purposes. It limits storage growth (inactive users' features expire automatically), and it acts as a staleness indicator. If a feature request returns None for all streaming features, the model knows the streaming pipeline hasn't computed features for this user recently and can fall back to batch features or default values.
We use a layered feature retrieval strategy at inference time: check the streaming feature store first (freshest data), fall back to the batch feature store (slower but more complete), and use default values as a last resort. The serving application logs which layer provided each feature, so we can track the streaming feature coverage rate and identify users or segments where streaming features are consistently unavailable.
Testing Streaming Pipelines
Testing Flink jobs is harder than testing batch pipelines because time is an explicit part of the computation. A "count of events in the last 5 minutes" feature depends on event timestamps and watermark progression, which are difficult to simulate in unit tests.
We use Flink's MiniCluster for integration tests. The MiniCluster runs a full Flink environment in-process, with a configurable event-time clock that we advance manually. Test fixtures define a sequence of events with explicit timestamps, and assertions verify the output features at specific points in simulated time.
For each streaming feature, we maintain a batch implementation that computes the same value from historical data. The two implementations serve as cross-references — if the batch and streaming versions produce different results for the same input data and time window, one of them has a bug. This redundancy catches subtle issues like off-by-one errors in window boundaries or incorrect handling of late-arriving events that are hard to catch in unit tests alone.