Kafka: a Distributed Messaging System for Log Processing

• Many early systems for processing this kind of data relied on physically scraping log files off production servers for analysis. In recent years, several specialized distributed log aggregators have been built, including Facebook’s Scribe , Yahoo’s Data Highway , and Cloudera’s Flume. Those systems are primarily designed for collecting and loading the log data into a data warehouse or Hadoop for offline consumption. （一些现有的日志系统都是非实时系统，将数据导入到数据仓库或是Hadoop做离线处理）
• At LinkedIn (a social network site), we found that in addition to traditional offline analytics, we needed to support most of the real-time applications mentioned above with delays of no more than a few seconds.（kafka主要用来针对日志做实时分析和在线处理的）
  • On the one hand, Kafka is distributed and scalable, and offers high throughput. （分布式，可扩展，高吞吐的架构）
  • On the other hand, Kafka provides an API similar to a messaging system and allows applications to consume log events in real time.（提供类似消息系统的API）

• Traditional enterprise messaging systems have existed for a long time and often play a critical role as an event bus for processing asynchronous data flows. However, there are a few reasons why they tend not to be a good fit for log processing. （传统的消息系统不适合做日志处理）
  • First, there is a mismatch in features offered by enterprise systems. Those systems often focus on offering a rich set of delivery guarantees. Such delivery guarantees are often overkill for collecting log data. For instance, losing a few pageview events occasionally is certainly not the end of the world. Those unneeded features tend to increase the complexity of both the API and the underlying implementation of those systems.（企业系统提供了一些无用的特性，并且这些特性复杂了API以及底层实现）
  • Second, many systems do not focus as strongly on throughput as their primary design constraint.（不强调吞吐）
  • Third, those systems are weak in distributed support. There is no easy way to partition and store messages on multiple machines.（分布式支持不太好）
  • Finally, many messaging systems assume near immediate consumption of messages, so the queue of unconsumed messages is always fairly small. Their performance degrades significantly if messages are allowed to accumulate, as is the case for offline consumers such as data warehousing applications that do periodic large loads rather than continuous consumption. （不能够支持存储时间过长的历史数据）
• A number of specialized log aggregators have been built over the last few years.
  • Facebook uses a system called Scribe. Each frontend machine can send log data to a set of Scribe machines over sockets. Each Scribe machine aggregates the log entries and periodically dumps them to HDFS or an NFS device.
  • Yahoo’s data highway project has a similar dataflow. A set of machines aggregate events from the clients and roll out “minute” files, which are then added to HDFS.
  • Flume is a relatively new log aggregator developed by Cloudera. It supports extensible “pipes” and “sinks”, and makes streaming log data very flexible. It also has more integrated distributed support
  • However, most of those systems are built for consuming the log data offline, and often expose implementation details unnecessarily (e.g. “minute files”) to the consumer（只是比较适合离线处理）
  • Additionally, most of them use a “push” model in which the broker forwards data to consumers.（并且使用的都是push模型）
• At LinkedIn, we find the “pull” model more suitable for our applications since each consumer can retrieve the messages at the maximum rate it can sustain and avoid being flooded by messages pushed faster than it can handle. The pull model also makes it easy to rewind a consumer（使用pull模型能够由client来控制速率，并且能够很容易地由client做rewind）

• We first introduce the basic concepts in Kafka
  • A stream of messages of a particular type is defined by a topic.
  • A producer can publish messages to a topic.
  • The published messages are then stored at a set of servers called brokers.
  • A consumer can subscribe to one or more topics from the brokers, and consume the subscribed messages by pulling data from the brokers.
    • Each message stream provides an iterator interface over the continual stream of messages being produced. The consumer then iterates over every message in the stream and processes the payload of the message.
    • Unlike traditional iterators, the message stream iterator never terminates. If there are currently no more messages to consume, the iterator blocks until new messages are published to the topic.（以iterators方式提供消息）
    • We support both the point-topoint delivery model in which multiple consumers jointly consume a single copy of all messages in a topic, as well as the publish/subscribe model in which multiple consumers each retrieve its own copy of a topic.（多个consumer可以共同消费一份，或者是各自消费一份）
• Since Kafka is distributed in nature, an Kafka cluster typically consists of multiple brokers. To balance load, a topic is divided into multiple partitions and each broker stores one or more of those partitions. Multiple producers and consumers can publish and retrieve messages at the same time.(一个topic被分割成多个partition, 每个broker上面可以host多个partition)

Producer Code

producer = new Producer(...);
message = new Message(“test message str”.getBytes());
set = new MessageSet(message);
producer.send(“topic1”, set);

Consumer Code

streams[] = Consumer.createMessageStreams(“topic1”, 1);
for (message : streams[0]) {
  bytes = message.payload();
  // do something with the bytes
}

• We rely on a hardware load-balancer to distribute the publish requests to the set of Kafka brokers evenly. （硬件负载均衡）
• We also deploy a cluster of Kafka in a separate datacenter for offline analysis, located geographically close to our Hadoop cluster and other data warehouse infrastructure. Without too much tuning, the end-to-end latency for the complete pipeline is about 10 seconds on average, good enough for our requirements.（跨机房数据延迟在10s以内）
• Our tracking also includes an auditing system to verify that there is no data loss along the whole pipeline.（检验数据是否丢失）
  • To facilitate that, each message carries the timestamp and the server name when they are generated. We instrument each producer such that it periodically generates a monitoring event, which records the number of messages published by that producer for each topic within a fixed time window.
  • The producer publishes the monitoring events to Kafka in a separate topic. The consumers can then count the number of messages that they have received from a given topic and validate those counts with the monitoring events to validate the correctness of data.
• Loading into the Hadoop cluster is accomplished by implementing a special Kafka input format that allows MapReduce jobs to directly read data from Kafka.

There are a number of directions that we’d like to pursue in the future.

• First, we plan to add built-in replication of messages across multiple brokers to allow durability and data availability guarantees even in the case of unrecoverable machine failures.(broker replicaiton需要线上，这样可以确保durability以及availability) We’d like to support both asynchronous and synchronous replication models to allow some tradeoff between producer latency and the strength of the guarantees provided. An application can choose the right level of redundancy based on its requirement on durability, availability and throughput. (在replication上面可以选择同步还是异步方式)
• Second, we want to add some stream processing capability in Kafka.（提供一些流式处理方面的能力）
  • After retrieving messages from Kafka, real time applications often perform similar operations such as window-based counting and joining each message with records in a secondary store or with messages in another stream.
  • At the lowest level this is supported by semantically partitioning messages on the join key during publishing so that all messages sent with a particular key go to the same partition and hence arrive at a single consumer process. This provides the foundation for processing distributed streams across a cluster of consumer machines.
  • On top of this we feel a library of helpful stream utilities, such as different windowing functions or join techniques will be beneficial to this kind of applications.

据说从0.8开始支持replication. 粗略地阅读了一下主页上的 documentation , 把一些比较关键东西记录下来。

除了replication之外，提供了两个高级功能：a. Consumer Offset Tracking（记录consumer当前消费位置） b. Log Compaction.（将日志文件中相同key的message进行合并压缩）

replication实现方式大致是这样的：

1. 在0.8之前一个topic/partition是由一个broker来进行管理的。这个broker就是master, 它没有任何followers
2. 在0.8之后每个topic/partition除了master之外还有followers. 生产者提交数据给主broker, 主broker写入数据后会将数据交给followers. 消费者也是从主brokers来拿数据的。所以这些slaves实际上都是inactive的。
3. producers写入数据时候可以指定说，是否写入master就返回，还是必须等待写入k个followers才返回，还是不需要等待master的ack就返回。策略牵扯到latency和durabilityz之间的tradeoff.
4. 常见quorum要求，如果集群节点数量为2f+1的话，那么必须得到f+1个节点的相同认定（容忍f个节点失效）。但是kafka quorum不是这样做的，kafka会在保存一个ISR(in-sync replicas)到可能是zookeeper上，这个集合代表当前有哪些节点是处于in-sync状态的。在进行quorum时候只允许ISR里面节点进行投票。和微软 PacificA 一致性协议比较类似，和常见quorum协议差别就在于在更新状态时候保存了ISR。如果节点数量为2f+1的话，那么可以容忍2f个节点失效。
5. 如果一个broker挂掉的话，迁移决策过程是由另外一个broker来代理完成的，这个broker称为controller.