At an abstract level, the dataflow systems discussed in the previous section give you a process for creating derived datasets (such as search indexes, materialized views, and predictive models) and keeping them up-to-date. Let’s call that process the write path. Whenever a piece of information is written to the system, it may go through multiple stages of batch and stream processing, and eventually every derived dataset is updated to incorporate the data that was written. Figure 13-1 shows an example of updating a search index.
Diagram illustrating the write and read paths in a search index, showing document updates processed through linguistic analysis to update the index, and queries accessing the index for search results.
Figure 13-1. In a search index, writes (document updates) meet reads (queries).
But why do you create the derived dataset in the first place? Most likely because you want to query it again at a later time. This is the read path: when serving a user request, you read from the derived dataset, perhaps perform more processing on the results, and construct the response to the user.
Taken together, the write path and the read path encompass the whole journey of the data, from the point where it is collected to the point where it is consumed (probably by another human). The write path is the portion of the journey that is precomputed; it’s done eagerly as soon as the data comes in, regardless of whether anyone has asked to see it. The read path is the portion of the journey that happens only when someone asks for it. If you are familiar with functional programming languages, you might notice that the write path is similar to eager evaluation and the read path is similar to lazy evaluation.
The derived dataset is the place where the write path and the read path meet, as illustrated in Figure 13-1. It represents a trade-off between the amount of work that needs to be done at write time and the amount that needs to be done at read time.
Materialized views and caching
A full-text search index is a good example: the write path updates the index, and the read path searches the index for keywords. Both reads and writes need to do some work. Writes need to update the index entries for all terms that appear in the document. Reads need to search for each word in the query and apply Boolean logic to find documents that contain all the words in the query (an AND operator) or any synonym of each word (an OR operator).
If you didn’t have an index, a search query would have to scan over all documents (like grep), which would get very expensive if you had a large number of documents. No index means less work on the write path (no index to update) but a lot more work on the read path.
On the other hand, you could imagine precomputing the search results for all possible queries. In that case, you would have less work to do on the read path: no Boolean logic, just find the results for your query and return them. However, the write path would be a lot more expensive. The set of possible search queries that could be asked is infinite (or at least exponential in the number of terms in the corpus), and thus precomputing all possible search results would not be possible.
Another option would be to precompute the search results for only a fixed set of the most common queries, so that they can be served quickly without having to go to the index. The uncommon queries can still be served from the index. This would generally be called a cache of common queries, although we could also call it a materialized view, as it would need to be updated when new documents appear that should be included in the results of one of the common queries.
From this example, we can see that an index is not the only possible boundary between the write path and the read path. Caching of common search results is possible, and grep-like scanning without the index is also possible on a small number of documents. Viewed like this, the role of caches, indexes, and materialized views is simple: they shift the boundary between the read path and the write path. They allow us to do more work on the write path, by precomputing results, in order to save effort on the read path.
Shifting the boundary between work done on the write path and the read path was in fact the topic of the social networking example in “Case Study: Social Network Home Timelines”. In that example, we also saw how the boundary between write path and read path might be drawn differently for celebrities compared to ordinary users. After 500 pages, we have come full circle!
Stateful, offline-capable clients
The idea of a boundary between write and read paths is interesting because we can discuss shifting that boundary and explore what that shift means in practical terms. Let’s consider it in a different context.
In the past, web browsers were stateless clients that could do useful things only when you had an internet connection (just about the only thing you could do offline was scroll up and down in a page that you had previously loaded while online). However, single-page JavaScript web apps now have a lot of stateful capabilities, including client-side UI interaction and persistent local storage in the web browser. Mobile apps can similarly store a lot of state on the device and don’t require a round trip to the server for most user interactions.
In “Sync Engines and Local-First Software” we saw how persistent local state enables a class of applications in which users can work offline, without an internet connection, and sync with remote servers in the background when a network connection is available. Since mobile devices sometimes have slow and unreliable cellular internet connections, it’s a big advantage for users if the UI doesn’t have to wait for synchronous network requests and apps mostly work offline.
When we move away from the assumption of stateless clients talking to a central database and toward state that is maintained on end-user devices, a world of new opportunities opens up. In particular, we can think of the on-device state as a cache of state on the server. The pixels on the screen are a materialized view of model objects in the client app; the model objects are a local replica of state in a remote datacenter.
Pushing state changes to clients
If you load a typical web page in a web browser and the data subsequently changes on the server, the browser does not find out about the change until you reload the page. The browser reads the data at only one point in time, assuming that it is static; the browser does not subscribe to updates from the server. Thus, the state in the browser is a stale cache that is not updated unless you explicitly poll for changes. (HTTP-based feed subscription protocols like RSS are really just a basic form of polling.)
More recent protocols have moved beyond the basic request/response pattern of HTTP. Server-sent events (the EventSource API) and WebSockets provide communication channels by which a web browser can keep an open TCP connection to a server and the server can actively push messages to the browser as long as it remains connected. This provides an opportunity for the server to actively inform the end-user client about any changes to the state it has stored locally, reducing the staleness of the client-side state.
In terms of our model of write path and read path, actively pushing state changes all the way to client devices means extending the write path all the way to the end user. When a client is first initialized, it will still need to use a read path to get its initial state, but thereafter it can rely on a stream of state changes sent by the server. The ideas we discussed around stream processing and messaging are thus not restricted to running in a datacenter; we can take them further and extend them all the way to end-user devices.
The devices will be offline some of the time, and they will be unable to receive any notifications of state changes from the server during that time. But we already solved that problem; in “Consumer offsets” we discussed how a consumer of a log-based message broker can reconnect after failing or becoming disconnected and ensure that it doesn’t miss any messages that arrived while it was disconnected. The same technique works for individual users, where each device is a small subscriber to a small stream of events.
End-to-end event streams
Tools for developing stateful clients and UIs, such as React and Elm, already have the ability to update the rendered UI in response to changes in the underlying state. It would be very natural to extend this programming model to also allow a server to push state change events into this client-side event pipeline.
State changes could then could flow through an end-to-end write path: from the interaction on one device that triggers the change, through event logs and various derived data systems and stream processors, all the way to the user interface on another device. These state changes could be propagated with fairly low delay—say, under one second end to end.
Some applications, such as instant messaging and online games, already have such a “real-time” architecture (in the sense of interactions with low delay, not in the sense of response time guarantees). Why don’t we build all applications this way?
The challenge is that the assumption of stateless clients and request/response interactions is deeply ingrained in our databases, libraries, frameworks, and protocols. Many datastores support read and write operations where a request returns a single response, but far fewer support operations where a request returns a stream of responses over time (i.e., the ability to subscribe to changes).
To extend the write path all the way to the end user, we would need to fundamentally rethink the way we build many of these systems, moving away from request/response interaction and toward publish/subscribe dataflow. This would require effort, but it would have the advantage of making UIs more responsive and providing better offline support.
Reads are events too
We discussed that when a stream processor writes derived data to a store (database, cache, or index) and that store is queried, the store acts as the boundary between the write path and the read path. It allows random-access read queries to the data that would otherwise require scanning the whole event log.
In many cases, the datastore is separate from the streaming system. However, recall that stream processors also need to maintain state to perform aggregations and joins. This state is normally hidden inside the stream processor, but some frameworks allow it to be queried by outside clients, turning the stream processor itself into a kind of simple database.
Let’s take that idea further. In the model we’ve discussed so far, writes to the store go through an event log, while reads are transient network requests that go directly to the nodes that store the data being queried. This is a reasonable design, but not the only possible one. It is also possible to represent read requests as streams of events and send both the read events and the write events through a stream processor. The processor responds to read events by emitting the result of the read to an output stream.
When both writes and reads are represented as events and routed to the same stream operator, we are in fact performing a stream-table join between the stream of read queries and the database. Each read event needs to be sent to the database shard holding the relevant data, just as batch and stream processors copartition inputs on the same key when performing joins.
This correspondence between serving requests and performing joins is quite fundamental. A one-off read request passes through the join operator, which then immediately forgets the request; a subscribe request is a persistent join with past and future events on the other side of the join.
Recording a log of read events potentially also has benefits with regard to tracking causal dependencies and data provenance across a system. The log would allow you to reconstruct what the user saw before they made a particular decision. For example, in an online shop, the predicted shipping date and the inventory status shown to a customer likely affect whether they choose to buy an item. To analyze this connection, you need to record the result of the user’s query of the shipping and inventory status.
Writing read requests to durable storage thus enables better tracking of causal relationships, but it incurs additional storage and I/O costs. Optimizing such systems to reduce the overhead is still an open research problem, but if you already log read requests for operational purposes as a side effect of request processing, it’s not a big change to make that log the source of the requests instead.
Multishard data processing
For queries that touch only a single shard, the effort of sending them through a stream and collecting a stream of responses is perhaps overkill. However, this idea opens the possibility of distributed execution of complex queries that need to combine data from several shards, taking advantage of the infrastructure for message routing, sharding, and joining that is already provided by stream processors.
Storm’s distributed RPC feature supports this usage pattern. For example, it has been used to compute the number of people who have seen a URL on a social network—that is, the union of the follower sets of everyone who has posted that URL. As the set of users is sharded, this computation requires combining results from many shards.
Another example of this pattern occurs in fraud prevention. To assess the risk of whether a particular purchase event is fraudulent, you can examine the reputation scores of the user’s IP address, email address, billing address, shipping address, and so on. Each of these reputation databases is itself sharded, so collecting the scores for a particular purchase event requires a sequence of joins with differently sharded datasets.
The internal query execution graphs of data warehouse query engines have similar characteristics. If you need to perform this kind of multishard join, it is probably simpler to use a database that provides this feature than to implement it using a stream processor. However, treating queries as streams provides an option for implementing large-scale applications that run against the limits of conventional off-the-shelf solutions.