In “Joins and Grouping” we discussed how batch jobs can join datasets by key and how such joins form an important part of data pipelines. Since stream processing generalizes data pipelines to incremental processing of unbounded datasets, there is exactly the same need for joins on streams.
However, the fact that new events can appear at any time on a stream makes joins on streams more challenging than in batch jobs. To understand the situation better, let’s distinguish between three types of joins: stream–stream joins, stream–table joins, and table–table joins. In the following sections we’ll illustrate each by example.
Stream–stream join (window join)
Say you have a search feature on your website, and you want to detect recent trends in searched-for URLs. Every time someone types a search query, you log an event containing the query and the results returned. Every time someone clicks one of the search results, you log another event recording the click. To calculate the click-through rate for each URL in the search results, you need to bring together the events for the search action and the click action, which are connected by having the same session ID. Similar analyses are needed in advertising systems.
The click may never come if the user abandons their search, and even if it comes, the time between the search and the click may be highly variable. In many cases it might be a few seconds, but it could be as long as days or weeks (if a user runs a search, forgets about that browser tab, and then returns to the tab and clicks a result sometime later). Because of variable network delays, the click event may even arrive before the search event. You can choose a suitable window for the join—for example, you may choose to join a click with a search if they occur at most one hour apart.
Note that embedding the details of the search in the click event is not equivalent to joining the events. Doing so would tell you only about the cases where the user clicked a search result, not about the searches where the user did not click any of the results. To measure search quality, you need accurate click-through rates, for which you need both the search events and the click events.
To implement this type of join, a stream processor needs to maintain state—for example, all the events that occurred in the last hour, indexed by session ID. Whenever a search event or click event occurs, it is added to the appropriate index, and the stream processor also checks the other index to see whether another event for the same session ID has already arrived. If there is a matching event, you emit an event saying which search result was clicked. If the search event expires without you seeing a matching click event, you emit an event saying which search results were not clicked.
Stream–table join (stream enrichment)
In “Joins and Grouping” (Figure 11-2) we saw an example of a batch job joining two datasets: a set of user activity events and a database of user profiles. It is natural to think of the user activity events as a stream and to perform the same join on a continuous basis in a stream processor. The input is a stream of activity events containing a user ID, and the output is a stream of activity events in which the user ID has been augmented with profile information about the user. This process is sometimes known as enriching the activity events with information from the database.
To perform this join, the stream process needs to take one activity event at a time, look up the event’s user ID in the database, and add the profile information to the activity event. The database lookup could be implemented by querying a remote database; however, as discussed in “Joins and Grouping”, such remote queries are likely to be slow and risk overloading the database.
Another approach is to load a copy of the database into the stream processor so that it can be queried locally without a network round trip. This technique is called a hash join since the local copy of the database might be an in-memory hash table if it is small enough, or an index on the local disk.
The difference from batch jobs is that a batch job uses a point-in-time snapshot of the database as input, whereas a stream processor is long-running, and the contents of the database are likely to change over time, so the stream processor’s local copy of the database needs to be kept up-to-date. This issue can be solved by CDC. The stream processor can subscribe to a changelog of the user profile database as well as the stream of activity events. When a profile is created or modified, the stream processor updates its local copy. Thus, we obtain a join between two streams: the activity events and the profile updates.
A stream–table join is very similar to a stream–stream join. The biggest difference is that for the table changelog stream, the join uses a window that reaches back to the “beginning of time” (a conceptually infinite window), with newer versions of records overwriting older ones. For the stream input, the join might not maintain a window at all.
Table–table join (materialized view maintenance)
Consider the social network timeline example that we discussed in “Case Study: Social Network Home Timelines”. We said that when a user wants to view their home timeline, it is too expensive to iterate over all the people the user is following, find their recent posts, and merge them.
Instead, we want a timeline cache, a kind of per-user “inbox” to which posts are written as they are sent, so that reading the timeline is a single lookup. Materializing and maintaining this cache requires the following event processing:
-
When user u sends a new post, it is added to the timeline of every user who is following u.
-
When a user deletes a post, or deletes their entire account, it is removed from all users’ timelines.
-
When user u1 starts following user u2, recent posts by u2 are added to u1’s timeline.
-
When user u1 unfollows user u2, posts by u2 are removed from u1’s timeline.
To implement this cache maintenance in a stream processor, you need streams of events for posts (sending and deleting) and for follow relationships (following and unfollowing). The stream process needs to maintain a database containing the set of followers for each user so that it knows which timelines need to be updated when a new post arrives.
Another way of looking at this stream process is that it maintains a materialized view for a query that joins two tables (posts and follows)—something like the following:
SELECT follows.follower_id AS timeline_id,
array_agg(posts.* ORDER BY posts.timestamp DESC)
FROM posts
JOIN follows ON follows.followee_id = posts.sender_id
GROUP BY follows.follower_id
The join of the streams corresponds directly to the join of the tables in this query. The timelines are effectively a cache of the result of the query, updated every time the underlying tables change.
Note
If you regard a stream as the derivative of a table, as in Figure 12-7, and regard a join as a product of two tables u·v, something interesting happens: the stream of changes to the materialized join follows the product rule (u·v)′ = u′v + uv′. Any change of posts is joined with the current followers, and any change of follows is joined with the current posts.
Time dependence of joins
The three types of joins described here (stream–stream, stream–table, and table–table) have a lot in common. They all require the stream processor to maintain a state (search and click events, user profiles, or follower list) derived from one join input and to query that state when processing records from the other input.
The order of the events that maintain the state is important—for example, it matters whether you first follow and then unfollow a user, or the other way round. In a sharded event log like Kafka, the ordering of events within a single shard (partition) is preserved, but there is typically no ordering guarantee across different streams or shards.
This raises a question: if events on different streams happen around a similar time, in which order are they processed? In the stream–table join example, for instance, if a user updates their profile, which activity events are joined with the old profile (processed before the profile update), and which are joined with the new profile (processed after the profile update)? Put another way: if state changes over time, and you join with a state, what point in time do you use for the join?
Such time dependence can occur in many places. For example, if you sell things, you need to apply the right tax rate to invoices, which depends on the country or state, the type of product, and the date of sale (since tax rates change from time to time). When joining sales to a table of tax rates, you probably want to join with the tax rate at the time of the sale, which may be different from the current tax rate if you are reprocessing historical data.
If the ordering of events across streams is undetermined, the join becomes nondeterministic, which means you cannot rerun the same job on the same input and necessarily get the same result. The events on the input streams may be interleaved in a different way when you run the job again.
In data warehouses, this issue is known as a slowly changing dimension (SCD), and it is often addressed by using a unique identifier for a particular version of the joined record—for example, every time the tax rate changes, it is given a new identifier, and the invoice includes the identifier for the tax rate at the time of sale. This change makes the join deterministic, but it has the consequence that log compaction is not possible, since all versions of the records in the table need to be retained. Alternatively, you can denormalize the data and include the applicable tax rate directly in every sale event.