LinkedIn: the log and the birth of Kafka

01 · CONTEXT

The problem that came before Kafka

Jay Kreps tells how LinkedIn started hitting the limits of a monolithic, centralized database and had to migrate to a set of specialized distributed systems. Among them were a distributed graph database, a distributed search engine, a Hadoop installation, and first- and second-generation key-value stores.

The Kafka paper describes the other side of the same problem: log and activity data. Older systems scraped log files directly from production servers, and the newer log aggregators were built mostly to load this data into a data warehouse or Hadoop, for offline consumption. At LinkedIn, besides offline analysis, most applications had to be served in real time with delays of at most a few seconds.

The author says he got involved in this problem around 2008, after they launched the key-value store. The next project was to build a working Hadoop and move some of the recommendation processes onto it, and the initial plan was to extract the data from the existing Oracle data warehouse.

1. The initial plan was to extract the data from the existing Oracle data warehouse.

2. The warehouse's processing was not suited to the production batch workload planned for Hadoop.

3. The team ended up bypassing the warehouse and going directly to the source databases and log files.

4. Finally, they built another pipeline to load data into the key-value store that serves the results.

5. Whenever any of the pipelines had a problem, Hadoop was largely useless.

02 · THE IDEA

One central log instead of custom pipelines

Even with the code written in a reasonably generic way, each new data source required its own configuration and proved to be a major source of errors and failures. There were dozens of data systems, and connecting all of them would lead to custom plumbing between every pair of systems.

Since many systems are both source and destination, there would be two data pipelines per system, one for input and one for output. Near full connectivity, the author estimates something like O(N2) pipelines.

The way out he describes is to isolate each consumer from the data source: a new system, whether source or destination, should require integration with only a single pipeline, not with each data consumer. That experience is what led to the focus on building Kafka, combining what they had seen in messaging systems with the concept of a log, popular in databases and inside distributed systems.

The log is perhaps the simplest possible storage abstraction: an append-only sequence of records, totally ordered by time. The post's recipe is to take the organization's data and put it into a central log for real-time subscription.

1. Each logical source can be modeled as its own log, whether an application that records events or a database table that accepts modifications.

2. Each subscriber system reads from the log as fast as it can, applies each new record to its own storage, and advances its position in the log.

3. A batch system, such as Hadoop or a data warehouse, may consume only hourly or daily, while a real-time query system may need the data up to the second.

4. A subscriber can go down or stop for maintenance and catch up when it returns, consuming at a pace it controls.

5. Neither the source nor the log knows the destinations, so consumer systems can be added and removed without changing the pipeline.

The author describes the log as a kind of mensageria with durability guarantees and strong ordering semantics.

03 · WHY NOT AN ORDINARY QUEUE

Traditional messaging was not the fit

The paper lists reasons why enterprise messaging is not a good fit for log data. These systems usually offer a rich set of delivery guarantees, and the paper considers these guarantees often overkill for collecting log data. Their performance also drops considerably when messages accumulate, as happens with offline consumers, for example data warehouse applications that do large periodic loads instead of continuous consumption.

The authors also prefer the pull model over push, in which the broker forwards data to the consumer: with pull, each consumer fetches messages at the maximum rate it can handle and avoids being flooded by messages pushed faster than it can process.

In Kafka, the information about how much each consumer has read is not kept on the broker, but on the consumer itself. The consumer can also deliberately rewind to an old offset and reconsume data, which violates the usual contract of a queue, but proved essential for many consumers. For example, if there is a bug in the consuming application's logic, it can reprocess certain messages after fixing it, something particularly important for ETL workloads into the data warehouse or Hadoop.

04 · EVENTS AND SCALE

From use in events to volume

The post shows the effect on the code. The job posting page now only displays the posting and records that it was shown, with attributes of the posting, of who viewed it, and other useful facts. The other interested systems, such as recommendations, security, advertiser analytics, and the data warehouse, simply subscribe to the feed and do their own processing.

At LinkedIn, Kafka is used as the central, multi-subscriber event log, with several hundred event types defined. According to the post, at the time they were running more than 60 billion unique message writes per day, and several hundred billion if you count the writes from mirroring between datacenters.

To scale horizontally, the log is split into partitions. Each partition is a totally ordered log, with no global order across partitions, and the writer controls the partition of each message, with most users partitioning by some key, such as the user id. Partitioning allows appending to the log without coordination between shards and makes throughput grow linearly with cluster size.

Each partition is replicated across a configurable number of replicas, each with an identical copy of the partition's log. One of them acts as leader and, if the leader fails, another replica takes over. The guarantee is that each partition preserves order and that appends from a single sender to a partition are delivered in the order sent; the lack of global order is a limitation, but the author says he did not find it serious.

The log also groups small reads and writes into larger operations. Batching occurs from client to server, in disk writes, in replication between servers, in transfer to consumers, and in data acknowledgment. A simple binary format is kept across the in-memory log, the on-disk log, and the network transfer, which enables optimizations such as zero-copy transfer. The cumulative effect is that you can normally write and read at the rate the disk or network supports, even with datasets much larger than memory.

05 · DEPLOYMENT

How the paper describes the deployment

The paper describes a Kafka cluster co-located with each datacenter where the user-facing services run.

1. The frontend services publish log data in batches to the local Kafka brokers, and a hardware load balancer distributes publish requests among the brokers.

2. The online Kafka consumers run in services inside the same datacenter.

3. A Kafka cluster in a separate datacenter, for offline analysis, runs embedded consumers that pull data from the Kafka clusters of the live datacenters.

4. Load jobs pull the data from this replica cluster into Hadoop and the data warehouse, and the Hadoop load uses a special input format that lets MapReduce jobs read directly from Kafka.

5. Without much tuning, the end-to-end latency of the complete pipeline was about 10 seconds on average, enough for their requirements.

The paper records that Kafka accumulated hundreds of gigabytes of data and close to a billion messages per day. There is also an auditing system to verify that no data is lost along the entire pipeline, and a schema that makes it possible to enforce a compatibility contract between producers and consumers.

06 · FROM PAPER TO POST

What changed between the two sources

The paper is dated June 12, 2011, and Jay Kreps's post is from December 16, 2013. The table compares what each source says about the same system.