Streaming Search Data Is a Time Problem
Search relevance teams have traditionally treated clickstream analysis as a batch job: process historical logs overnight, then update rankings the next day. But that approach can't react to what users are doing right now. Shopify's Discovery team is taking a different path, using streaming systems that can consume both live and historical data to auto-tune results in near real-time—boosting documents users click, demoting ones they ignore, and even building offline evaluation sets for learning-to-rank models.
The catch is that streaming systems introduce a fundamental complication: time. Events arrive late, out of order, or both. Apache Beam, the unified batch-and-stream processing framework Shopify uses for these workflows, gives you tools to handle that complexity. But before you can use them, you have to understand what Beam actually does with time.
Why Beam Instead of Separate Systems?
Using Spark for historical replay and Storm for live traffic means maintaining two codebases that solve overlapping problems. Beam consolidates both into one programming model. For search, that means you can write a pipeline that computes aggregate click popularity from months of logs, then apply the same logic to a live stream of user behavior without duplicating the implementation.
Beam pipelines connect a source through processing steps to a sink. Each event flowing through carries a timestamp, but those timestamps reflect when events were generated, not when they arrive at your pipeline. Network delays, API buffering, and intermediate stores all inject latency and disorder into the stream. Kafka sources in Beam, for example, maintain a watermark—a moving estimate of how late events can arrive—to emit data in as close to original order as possible.
Getting the timestamp configuration right at the source is the first milestone. If you don't, events get dropped or processed in the wrong sequence. And if you're joining multiple data streams—say, search queries with click events—the pipeline must align watermark progress across sources before deciding how long to wait for stragglers.
Setting Up a Timestamp Policy
A basic Kafka source in Java requires just a topic, server addresses, and serializers:
That pipeline produces SearchQueryEvent objects, but by default they're timestamped with Kafka processing time—the moment the event was read from the topic. That's rarely what matters for search. You want to know when a user actually typed a query or clicked a result, not when the event happened to be consumed.
Kafka sources can use the record's embedded create time instead, with a watermark allowance for how out-of-order those timestamps may be:
.withCreateTime(Duration.standardMinutes(5))
The five-minute value tells Beam: don't wait longer than this much event-time discrepancy before treating the stream as caught up. But Beam sources have no inherent knowledge of your domain's data model. If your event payload contains its own timestamp, you can define custom logic for extracting it.
For a SearchQueryEvent containing a searchTimestamp field, you provide a function that maps each record (a key-value of Long to SearchQueryEvent) to an Instant. That function is wrapped in a factory class—something like SearchQueryTimestampPolicyFactory—and attached to the source builder:
.withTimestampPolicyFactory(new SearchQueryTimestampPolicyFactory())
This gives Beam full control over how event time is derived from your own data, while still enforcing the same allowed delay for late arrivals.
Event Time vs. Processing Time
Beam operates on two distinct clocks. Event time is the timestamp that matters to your domain—when a search happened, when a movie scene took place. Processing time is the wall-clock reality of your pipeline execution. These can diverge wildly, and Beam's framework must reconcile the two.
To make that concrete, imagine watching The Goonies at 1000x speed, like Lieutenant Commander Data on Star Trek. Data receives the film's frames faster than real time, but the Enterprise computer occasionally delivers frames out of sequence. In that scenario:
- Event time is the movie's runtime—the 1 hour 55 minutes of plot that unfolds on screen.
- Processing time is how long Data actually spends experiencing it—perhaps just a few minutes of his day.
When Data tells the computer to tolerate up to five minutes of movie-time delay before showing what's available, he's accepting that some frames will be dropped. His experience remains coherent even if a handful of frames are missing. That's exactly how Beam behaves with streaming data. If the source says "wait for up to five minutes of event time," that may mean five real-time minutes when processing a live stream, or it could be milliseconds when replaying historical data where nothing actually arrives late—because, in event-time terms, the window has already elapsed.
When you use event-time-based processing, writing "withCreateTime(Duration.standardMinutes(5))" does not instruct the computer to pause execution for five minutes. The replay of historical data compresses time dramatically. Only for genuinely live streams does that delay correspond to what we'd intuitively call waiting.
Windows and Late-Data Handling
Beyond the source, Beam's Window transforms give you finer control over how data is buffered and aggregated over event-time units. Suppose you collect search data in daily windows: Beam must determine when a day's window closes, which depends on how the watermark advances and how much tolerance you've built in for late arrivals. If your downstream processing expects complete counts per query for a given period, windows must respect the same time-machine semantics as the source timestamp policies.
The right delay for a pipeline isn't always five minutes. It depends on how much latency your use case can tolerate and how much late data your downstream consumers can absorb. But once you accept that Beam's time model is fundamentally about event order—not execution speed—you can reason about these tradeoffs deliberately rather than by guesswork.
