语义分区

对Hadoop以及其他批量数据装载的支持

具有伸缩性的持久化方案使得Kafka可支持批量数据装载，可以周期性将快照数据加载进行批量处理的离线系统。

我们利用这个功能将数据加载我们的数据仓库（data warehouse）和Hadoop集群。

批量处理始于数据加载阶段。然后进入非循环图（acyclic graph）处理过程以及输出阶段（支持情况在这里）。

支持这样的处理模型的一个重要特性是。要有又一次装载从某个时间点開始的数据的能力（以防处理中有不论什么发生错误）。

对于Hadoop，我们通过在单个的map任务之上切割装载任务对数据的装载进行了并行化处理，切割时，全部节点/话题/分区的每种组合都要分出一个来。Hadoop提供了任务管理，失败的任务能够重头再来，不存在数据被反复的危急。

实施细则

以下给出了一些在上一节所描写叙述的低层相关的实现系统的某些部分的细节的简要说明。

API 设计

生产者 APIs

生产者 API 是给两个底层生产者的再封装 -kafka.producer.SyncProducerandkafka.producer.async.AsyncProducer.

class Producer {
	
  /* Sends the data, partitioned by key to the topic using either the */
  /* synchronous or the asynchronous producer */
  public void send(kafka.javaapi.producer.ProducerData producerData);

  /* Sends a list of data, partitioned by key to the topic using either */
  /* the synchronous or the asynchronous producer */
  public void send(java.util.List< kafka.javaapi.producer.ProducerData> producerData);

  /* Closes the producer and cleans up */	
  public void close();

} 

该API的目的是将生产者的全部功能通过一个单个的API公开给其使用者（client）。

新建的生产者能够：

• 对多个生产者请求进行排队/缓冲并异步发送批量数据 —— kafka.producer.Producer提供了在将多个生产请求序列化并发送给适当的Kafka代理分区之前，对这些生产请求进行批量处理的能力（producer.type=async）。批量的大小能够通过一些配置參数进行控制。

  当事件进入队列时会先放入队列进行缓冲。直到时间到了queue.time或者批量大小到达batch.size为止，后台线程（kafka.producer.async.ProducerSendThread）会将这批数据从队列中取出，交给kafka.producer.EventHandler进行序列化并发送给适当的kafka代理分区。通过event.handler这个配置參数。能够在系统中插入一个自己定义的事件处理器。

  在该生产者队列管道中的各个不同阶段，为了插入自己定义的日志/跟踪代码或者自己定义的监视逻辑，如能注入回调函数会很实用。通过实现kafka.producer.asyn.CallbackHandler接口并将配置參数callback.handler设置为实现类就能够实现注入。

• 使用用户指定的Encoder处理数据的序列化（serialization）

interfaceEncoder<T> {
  publicMessage toMessage(T data);
}

• 
  
  Encoder的缺省值是一个什么活都不干的kafka.serializer.DefaultEncoder。
• 提供基于zookeeper的代理自己主动发现功能 —— 通过使用zk.connect配置參数指定zookeeper的连接url，就行使用基于zookeeper的代理发现和负载均衡功能。

  在有些应用场合，可能不太适合于依赖zookeeper。

  在这样的情况下。生产者可以从broker.list这个配置參数中获得一个代理的静态列表，每一个生产请求会被随即的分配给各代理分区。假设对应的代理宕机。那么生产请求就会失败。

• 通过使用一个可选性的、由用户指定的Partitioner，提供由软件实现的负载均衡功能 —— 数据发送路径选择决策受kafka.producer.Partitioner的影响。

interfacePartitioner<T> {
   intpartition(T key, intnumPartitions);
}

• 分区API依据相关的键值以及系统中具有的代理分区的数量返回一个分区id。将该id用作索引，在broker_id和partition组成的经过排序的列表中为对应的生产者请求找出一个代理分区。缺省的分区策略是hash(key)%numPartitions。

  假设key为null，那就进行随机选择。使用partitioner.class这个配置參数也能够插入自己定义的分区策略。

使用者API

我们有两个层次的使用者API。

底层比較简单的API维护了一个同单个代理建立的连接，全然同发送给server的网络请求相吻合。该API全然是无状态的，每一个请求都带有一个偏移量作为參数，从而同意用户以自己选择的随意方式维护该元数据。

高层API对使用者隐藏了代理的详细细节，让使用者可执行于集群中的机器之上而无需关心底层的拓扑结构。它还维护着数据使用的状态。高层API还提供了订阅同一个过滤表达式（比如，白名单或黑名单的正則表達式）相匹配的多个话题的能力。

底层API

class SimpleConsumer {
	
  /* Send fetch request to a broker and get back a set of messages. */ 
  public ByteBufferMessageSet fetch(FetchRequest request);

  /* Send a list of fetch requests to a broker and get back a response set. */ 
  public MultiFetchResponse multifetch(List<FetchRequest> fetches);

  /**
   * Get a list of valid offsets (up to maxSize) before the given time.
   * The result is a list of offsets, in descending order.
   * @param time: time in millisecs,
   *              if set to OffsetRequest$.MODULE$.LATIEST_TIME(), get from the latest offset available.
   *              if set to OffsetRequest$.MODULE$.EARLIEST_TIME(), get from the earliest offset available.
   */
  public long[] getOffsetsBefore(String topic, int partition, long time, int maxNumOffsets);
}

底层API不但用于实现高层API，并且还直接用于我们的离线使用者（比方Hadoop这个使用者），这些使用者还对状态的维护有比較特定的需求。

高层API

/* create a connection to the cluster */ 
ConsumerConnector connector = Consumer.create(consumerConfig);

interface ConsumerConnector {
	
  /**
   * This method is used to get a list of KafkaStreams, which are iterators over
   * MessageAndMetadata objects from which you can obtain messages and their
   * associated metadata (currently only topic).
   *  Input: a map of <topic, #streams>
   *  Output: a map of <topic, list of message streams>
   */
  public Map<String,List<KafkaStream>> createMessageStreams(Map<String,Int> topicCountMap); 

  /**
   * You can also obtain a list of KafkaStreams, that iterate over messages
   * from topics that match a TopicFilter. (A TopicFilter encapsulates a
   * whitelist or a blacklist which is a standard Java regex.)
   */
  public List<KafkaStream> createMessageStreamsByFilter(
      TopicFilter topicFilter, int numStreams);

  /* Commit the offsets of all messages consumed so far. */
  public commitOffsets()
  
  /* Shut down the connector */
  public shutdown()
}

该API的中心是一个由KafkaStream这个类实现的迭代器（iterator）。每一个KafkaStream都代表着一个从一个或多个分区到一个或多个server的消息流。每一个流都是使用单个线程进行处理的。所以，该API的使用者在该API的创建调用中能够提供所需的随意个数的流。这样，一个流可能会代表多个server分区的合并（同处理线程的数目同样）。但每一个分区仅仅会把数据发送给一个流中。

createMessageStreams方法为使用者注冊到对应的话题之上，这将导致须要对使用者/代理的分配情况进行又一次平衡。为了将又一次平衡操作降低到最小。

该API鼓舞在一次调用中就创建多个话题流。createMessageStreamsByFilter方法为发现同其过滤条件想匹配的话题（额外地）注冊了多个监视器（watchers）。

应该注意。createMessageStreamsByFilter方法所返回的每一个流都可能会对多个话题进行迭代（比方，在满足过滤条件的话题有多个的情况下）。

网络层

消息的格式

/** 
	 * A message. The format of an N byte message is the following: 
	 * 
	 * If magic byte is 0 
	 * 
	 * 1. 1 byte "magic" identifier to allow format changes 
	 * 
	 * 2. 4 byte CRC32 of the payload 
	 * 
	 * 3. N - 5 byte payload 
	 * 
	 * If magic byte is 1 
	 * 
	 * 1. 1 byte "magic" identifier to allow format changes 
	 * 
	 * 2. 1 byte "attributes" identifier to allow annotations on the message independent of the version (e.g. compression enabled, type of codec used) 
	 * 
	 * 3. 4 byte CRC32 of the payload 
	 * 
	 * 4. N - 6 byte payload 
	 * 
	 */

日志

具有两个分区的、名称为"my_topic"的话题的日志由两个文件夹组成（即：my_topic_0和my_topic_1)，文件夹中存储的是内容为该话题的消息的数据文件。日志的文件格式是一系列的“日志项”；每条日志项包括一个表示消息长度的4字节整数N。其后接着保存的是N字节的消息。

每条消息用一个64位的整数偏移量进行唯一性标示。该偏移量表示了该消息在那个分区中的那个话题下发送的全部消息组成的消息流中所处的字节位置。每条消息在磁盘上的格式例如以下文所看到的。

每一个日志文件的以它所包括的第一条消息的偏移量来命名。因此，第一个创建出来的文件的名字将为00000000000.kafka，随后每一个后加的文件的名字将是前一个文件的文件名称大约再加S个字节所得的整数，当中，S是配置文件里指定的最大日志文件的大小。

消息的确切的二进制格式都有版本号。它保持为一个标准的接口，让消息集能够依据须要在生产者、代理、和使用者直接进行自由传输而无须又一次拷贝或转换。其格式例如以下所看到的：

On-disk format of a message

message length : 4 bytes (value: 1+4+n) 
"magic" value  : 1 byte
crc            : 4 bytes
payload        : n bytes

将消息的偏移量作为消息的可不常见。

我们原先的想法是使用由生产者产生的GUID作为消息id，然后在每一个代理上作一个从GUID到偏移量的映射。可是，既然使用者必须为每一个server维护一个ID，那么GUID所具有的全局唯一性就失去了价值。更有甚者，维护将从一个随机数到偏移量的映射关系带来的复杂性，使得我们必须使用一种重量级的索引结构，并且这样的结构还必须与磁盘保持同步，这样我们还就必须使用一种全然持久化的、需随机訪问的数据结构。如此一来，为了简化查询结构，我们就决定使用一个简单的依分区的原子计数器（atomic counter），这个计数器能够同分区id以及节点id结合起来唯一的指定一条消息；这样的方法使得查询结构简化不少，虽然每次在处理使用者请求时仍有可能会涉及多次磁盘寻道操作。然而，一旦我们决定使用计数器。跳向直接使用偏移量作为id就很自然了，毕竟两者都是分区内具有唯一性的、单调添加的整数。既然偏移量是在使用者API中并不会体现出来，所以这个决策终于还是属于一个实现细节，进而我们就选择了这样的更加高效的方式。

写操作

读操作

从某偏移量開始进行日志读取的实际过程须要先找出存储所需数据的日志段文件，从全局偏移量计算出文件内偏移量，然后再从该文件偏移量处開始读取。搜索过程通过对每一个文件保存在内存中的范围值进行一种变化后的二分查找完毕。

日志提供了获取最新写入的消息的功能，从而同意从“当下”開始消息订阅。

这个功能在使用者在SLA规定的天数内没能正常使用数据的情况下也非常实用。当使用者企图从一个并不存在的偏移量開始使用数据时就会出现这样的情况，此时使用者会得到一个OutOfRangeException异常，它能够依据详细的使用情况对自己进行重新启动或者只失败而退出。

下面是发送给数据使用者（consumer）的结果的格式。

MessageSetSend (fetch result)

total length     : 4 bytes
error code       : 2 bytes
message 1        : x bytes
...
message n        : x bytes

MultiMessageSetSend (multiFetch result)

total length       : 4 bytes
error code         : 2 bytes
messageSetSend 1
...
messageSetSend n

删除

一次仅仅能删除一个日志段的数据。 日志管理器同意通过可载入的删除策略设定删除的文件。 当前策略删除改动事件超过N 天以上的文件，也能够选择保留最后 N GB 的数据。 为了避免删除时的读取锁定冲突。我们能够使用副本写入模式，以便在进行删除的同一时候对日志段的一个不变的静态快照进行二进制搜索。

数据正确性保证

日志功能里有一个配置參数M，可对在强制进行磁盘刷新之前可写入的消息的最大条目数进行控制。在系统启动时会执行一个日志恢复过程，对最新的日志段内全部消息进行迭代。以对每条消息项的有效性进行验证。

一条消息项是合法的，仅当其大小加偏移量小于文件的大小而且该消息中有效载荷的CRC32值同该消息中存储的CRC值相等。在探測出有数据损坏的情况下，就要将文件依照最后一个有效的偏移量进行截断。

要注意。这里有两种必需处理的数据损坏情况：因为系统崩溃造成的未被正常写入的数据块（block）因而须要截断的情况以及因为文件里被增加了毫无意义的数据块而造成的数据损坏情况。造成数据损坏的原因是，一般来说OS并不能保证文件索引节点（inode）和实际数据块这两者的写入顺序，因此。除了可能会丢失未刷新的已写入数据之外，在索引节点已经用新的文件大小更新了但在将数据块写入磁盘块之前发生了系统崩溃的情况下。文件就可能会获得一些毫无意义的数据。CRC值就是用于这样的极端情况，避免由此造成整个日志文件的损坏（虽然未得到保存的消息当然是真的找不回来了）。

分发

Zookeeper文件夹

接下来讨论zookeeper用于在使用者和代理直接进行协调的结构和算法。

记法

当一个路径中的元素是用[xyz]这样的形式表示的时，其意思是, xyz的值并不固定并且实际上xyz的每种可能的值都有一个zookpeer z节点（znode）。比如。/topics/[topic]表示了一个名为/topics的文件夹。当中包括的子文件夹同话题相应。一个话题一个文件夹并且文件夹名即为话题的名称。也能够给出数字范围。比如[0...5]，表示的是子文件夹0、1、2、3、4。箭头->用于给出z节点的内容。比如/hello -> world表示的是一个名称为/hello的z节点，包括的值为"world"。

代理节点的注冊

/brokers/ids/[0...N] --> host:port (ephemeral node)

上面是全部出现的代理节点的列表，列表中每一项都提供了一个具有唯一性的逻辑代理id。用于让使用者可以识别代理的身份（这个必须在配置中给出）。在启动时，代理节点就要用/brokers/ids下列出的逻辑代理id创建一个z节点，并在自己注冊到系统中。

使用逻辑代理id的目的是，可以让我们在不影响数据使用者的情况下就能把一个代理搬到还有一台不同的物理机器上。试图用已在使用中的代理id（比方说，两个server配置成了同一个代理id）进行注冊会导致错误发生。

由于代理是以非长久性z节点的方式注冊的，所以这个注冊过程是动态的，当代理关闭或宕机后注冊信息就会消失（至此要数据使用者。该代理不再有效）。

代理话题的注冊

/brokers/topics/[topic]/[0...N] --> nPartions (ephemeral node)

每一个代理会都要注冊在某话题之下。注冊后它会维护并保存该话题的分区总数。

使用者和使用者小组

为了对数据的使用进行负载均衡并记录使用者使用的每一个代理上的每一个分区上的偏移量，全部话题的使用者都要在Zookeeper中进行注冊。

多个使用者能够组成一个小组共同使用一个单个的话题。同一小组内的每一个使用者共享同一个给定的group_id。比方说，假设某个使用者负责用三台机器进行某某处理过程。你就能够为这组使用者分配一个叫做“某某”的id。

这个小组id是在使用者的配置文件里指定的，而且这就是你告诉使用者它究竟属于哪个组的方法。

小组内的使用者要尽量公正地划分出分区，每一个分区仅为小组内的一个使用者所使用。

使用者ID的注冊

除了小组内的全部使用者都要共享一个group_id之外。每一个使用者为了要同其他使用者差别开来，还要有一个非永久性的、具有唯一性的consumer_id(採用hostname:uuid的形式）。 consumer_id要在下面的文件夹中进行注冊。

/consumers/[group_id]/ids/[consumer_id] --> {"topic1": #streams, ..., "topicN": #streams} (ephemeral node)

小组内的每一个使用者都要在它所属的小组中进行注冊并採用consumer_id创建一个z节点。

z节点的值包括了一个<topic, #streams>的map。 consumer_id仅仅是用来识别小组内活跃的每一个使用者。使用者建立的z节点是个暂时性的节点，因此假设这个使用者进程终止了，注冊信息也将随之消失。

数据使用者偏移追踪

数据使用者跟踪他们在每一个分区中耗用的最大偏移量。

这个值被存储在一个Zookeeper(分布式协调系统)文件夹中。

/consumers/[group_id]/offsets/[topic]/[broker_id-partition_id] --> offset_counter_value ((persistent node)

分区拥有者注冊表

每一个代理分区都被分配给了指定使用者小组中的单个数据使用者。数据使用者必须在耗用给定分区前确立对其的全部权。要确立其全部权，数据使用者须要将其 id 写入到特定代理分区中的一个暂时节点(ephemeral node)中。

/consumers/[group_id]/owners/[topic]/[broker_id-partition_id] --> consumer_node_id (ephemeral node)

代理节点的注冊

代理节点之间基本上都是相互独立的，因此它们仅仅须要公布它们拥有的信息。当有新的代理增加进来时，它会将自己注冊到代理节点注冊文件夹中，写下它的主机名和port。代理还要将已有话题的列表和它们的逻辑分区注冊到代理话题注冊表中。在代理上生成新话题时，须要动态的对话题进行注冊。

使用者注冊算法

当使用者启动时。它要做下面这些事情：

1. 将自己注冊到它属小组下的使用者id注冊表。
2. 注冊一个监视使用者id列的表变化情况（有新的使用者增加或者不论什么现有使用者的离开）的变化监视器。(每一个变化都会触发一次对发生变化的使用者所属的小组内的全部使用者进行负载均衡。)
3. 主次一个监视代理id注冊表的变化情况（有新的代理增加或者不论什么现有的代理的离开）的变化监视器。（每一个变化都会触发一次对全部小组内的全部使用者负载均衡。）
4. 假设使用者使用某话题过滤器创建了一个消息流，它还要注冊一个监视代理话题变化情况（增加了新话题）的变化监视器。（每一个变化都会触发一次对全部可用话题的评估，以找出话题过滤器过滤出哪些话题。新过滤出来的话题将触发一次对该使用者所在的小组内全部的使用者负载均衡。）
5. 迫使自己在小组内进行又一次负载均衡。

使用者又一次负载均衡的算法

使用者又一次复杂均衡的算法可用让小组内的全部使用者对哪个使用者使用哪些分区达成一致意见。使用者又一次负载均衡的动作每次加入或移除代理以及同一小组内的使用者时被触发。对于一个给定的话题和一个给定的使用者小组，代理分区是在小组内的全部使用者中进行平均划分的。一个分区总是由一个单个的使用者使用。

这样的设计方案简化了实施过程。如果我们执行多个使用者以并发的方式同一时候使用同一个分区，那么在该分区上就会形成争用（contention）的情况。这样一来就须要某种形式的锁定机制。

如果使用者的个数比分区多，就会出现有写使用者根本得不到数据的情况。在又一次进行负载均衡的过程中，我们依照尽量降低每一个使用者须要连接的代理的个数的方式，尝尝试着将分区分配给使用者。

每一个使用者在又一次进行负载均衡时须要做下列的事情：

   1.   针对Ci所订阅的每一个话题T
   2.   将PT设为生产话题T的全部分区
   3.   将CG设为小组内同Ci 一样使用话题T的全部使用者
   4.   对PT进行排序（让同一个代理上的各分区挨在一起）
   5.   对CG进行排序 
   6.   将i设为Ci在CG中的索引值并让N = size(PT)/size(CG)
   7.   将从i*N到(i+1)*N - 1的分区分配给使用者Ci 
   8.   将Ci当前所拥有的分区从分区拥有者注冊表中删除
   9.   将新分配的分区增加到分区拥有者注冊表中
        （我们可能须要多次尝试才干让原先的分区拥有者释放其拥有权）
       

在触发了一个使用者要又一次进行负载均衡时，同一小组内的其他使用者也会差点儿在同一时候被触发又一次进行负载均衡。