HBase数据模型

HBase简介

HBase – Hadoop Database，是一个高可靠性、高性能、面向列、可伸缩的分布式存储系统，利用HBase技术可在廉价PC Server上搭建起大规模结构化存储集群。

HBase是Google Bigtable的开源实现，类似Google Bigtable利用GFS作为其文件存储系统，HBase利用Hadoop HDFS作为其文件存储系统；Google运行MapReduce来处理Bigtable中的海量数据，HBase同样利用Hadoop MapReduce来处理HBase中的海量数据；Google Bigtable利用 Chubby作为协同服务，HBase利用Zookeeper作为对应。

上图描述了Hadoop EcoSystem中的各层系统，其中HBase位于结构化存储层，Hadoop HDFS为HBase提供了高可靠性的底层存储支持，Hadoop MapReduce为HBase提供了高性能的计算能力，Zookeeper为HBase提供了稳定服务和failover机制。

此外，Pig和Hive还为HBase提供了高层语言支持，使得在HBase上进行数据统计处理变的非常简单。 Sqoop则为HBase提供了方便的RDBMS数据导入功能，使得传统数据库数据向HBase中迁移变的非常方便。

HBase访问接口

1.       Native Java API，最常规和高效的访问方式，适合Hadoop MapReduce Job并行批处理HBase表数据

2.       HBase Shell，HBase的命令行工具，最简单的接口，适合HBase管理使用

3.       Thrift Gateway，利用Thrift序列化技术，支持C++，PHP，Python等多种语言，适合其他异构系统在线访问HBase表数据

4.       REST Gateway，支持REST 风格的Http API访问HBase, 解除了语言限制

5.       Pig，可以使用Pig Latin流式编程语言来操作HBase中的数据，和Hive类似，本质最终也是编译成MapReduce Job来处理HBase表数据，适合做数据统计

6.       Hive，当前Hive的Release版本尚没有加入对HBase的支持，但在下一个版本Hive 0.7.0中将会支持HBase，可以使用类似SQL语言来访问HBase

HBase数据模型

Table & Column Family

Ø  Row Key: 行键，Table的主键，Table中的记录按照Row Key排序

Ø  Timestamp: 时间戳，每次数据操作对应的时间戳，可以看作是数据的version number

Ø  Column Family：列簇，Table在水平方向有一个或者多个Column Family组成，一个Column Family中可以由任意多个Column组成，即Column Family支持动态扩展，无需预先定义Column的数量以及类型，所有Column均以二进制格式存储，用户需要自行进行类型转换。

Table & Region

当Table随着记录数不断增加而变大后，会逐渐分裂成多份splits，成为regions，一个region由[startkey,endkey)表示，不同的region会被Master分配给相应的RegionServer进行管理：

-ROOT- && .META. Table

HBase中有两张特殊的Table，-ROOT-和.META.

Ø  .META.：记录了用户表的Region信息，.META.可以有多个regoin

Ø  -ROOT-：记录了.META.表的Region信息，-ROOT-只有一个region

Ø  Zookeeper中记录了-ROOT-表的location

Client访问用户数据之前需要首先访问zookeeper，然后访问-ROOT-表，接着访问.META.表，最后才能找到用户数据的位置去访问，中间需要多次网络操作，不过client端会做cache缓存。

MapReduce on HBase

在HBase系统上运行批处理运算，最方便和实用的模型依然是MapReduce，如下图：

HBase Table和Region的关系，比较类似HDFS File和Block的关系，HBase提供了配套的TableInputFormat和TableOutputFormat API，可以方便的将HBase Table作为Hadoop MapReduce的Source和Sink，对于MapReduce Job应用开发人员来说，基本不需要关注HBase系统自身的细节。

HBase系统架构

Client

HBase Client使用HBase的RPC机制与HMaster和HRegionServer进行通信，对于管理类操作，Client与HMaster进行RPC；对于数据读写类操作，Client与HRegionServer进行RPC

Zookeeper

Zookeeper Quorum中除了存储了-ROOT-表的地址和HMaster的地址，HRegionServer也会把自己以Ephemeral方式注册到Zookeeper中，使得HMaster可以随时感知到各个HRegionServer的健康状态。此外，Zookeeper也避免了HMaster的单点问题，见下文描述

HMaster

HMaster没有单点问题，HBase中可以启动多个HMaster，通过Zookeeper的Master Election机制保证总有一个Master运行，HMaster在功能上主要负责Table和Region的管理工作：

1.       管理用户对Table的增、删、改、查操作

2.       管理HRegionServer的负载均衡，调整Region分布

3.       在Region Split后，负责新Region的分配

4.       在HRegionServer停机后，负责失效HRegionServer 上的Regions迁移

HRegionServer

HRegionServer主要负责响应用户I/O请求，向HDFS文件系统中读写数据，是HBase中最核心的模块。

HRegionServer内部管理了一系列HRegion对象，每个HRegion对应了Table中的一个Region，HRegion中由多个HStore组成。每个HStore对应了Table中的一个Column Family的存储，可以看出每个Column Family其实就是一个集中的存储单元，因此最好将具备共同IO特性的column放在一个Column Family中，这样最高效。

HStore存储是HBase存储的核心了，其中由两部分组成，一部分是MemStore，一部分是StoreFiles。MemStore是Sorted Memory Buffer，用户写入的数据首先会放入MemStore，当MemStore满了以后会Flush成一个StoreFile（底层实现是HFile），当StoreFile文件数量增长到一定阈值，会触发Compact合并操作，将多个StoreFiles合并成一个StoreFile，合并过程中会进行版本合并和数据删除，因此可以看出HBase其实只有增加数据，所有的更新和删除操作都是在后续的compact过程中进行的，这使得用户的写操作只要进入内存中就可以立即返回，保证了HBase I/O的高性能。当StoreFiles Compact后，会逐步形成越来越大的StoreFile，当单个StoreFile大小超过一定阈值后，会触发Split操作，同时把当前Region Split成2个Region，父Region会下线，新Split出的2个孩子Region会被HMaster分配到相应的HRegionServer上，使得原先1个Region的压力得以分流到2个Region上。下图描述了Compaction和Split的过程：

在理解了上述HStore的基本原理后，还必须了解一下HLog的功能，因为上述的HStore在系统正常工作的前提下是没有问题的，但是在分布式系统环境中，无法避免系统出错或者宕机，因此一旦HRegionServer意外退出，MemStore中的内存数据将会丢失，这就需要引入HLog了。每个HRegionServer中都有一个HLog对象，HLog是一个实现Write Ahead Log的类，在每次用户操作写入MemStore的同时，也会写一份数据到HLog文件中（HLog文件格式见后续），HLog文件定期会滚动出新的，并删除旧的文件（已持久化到StoreFile中的数据）。当HRegionServer意外终止后，HMaster会通过Zookeeper感知到，HMaster首先会处理遗留的 HLog文件，将其中不同Region的Log数据进行拆分，分别放到相应region的目录下，然后再将失效的region重新分配，领取 到这些region的HRegionServer在Load Region的过程中，会发现有历史HLog需要处理，因此会Replay HLog中的数据到MemStore中，然后flush到StoreFiles，完成数据恢复。

HBase存储格式

HBase中的所有数据文件都存储在Hadoop HDFS文件系统上，主要包括上述提出的两种文件类型：

1.       HFile， HBase中KeyValue数据的存储格式，HFile是Hadoop的二进制格式文件，实际上StoreFile就是对HFile做了轻量级包装，即StoreFile底层就是HFile

2.       HLog File，HBase中WAL（Write Ahead Log） 的存储格式，物理上是Hadoop的Sequence File

HFile

下图是HFile的存储格式：

首先HFile文件是不定长的，长度固定的只有其中的两块：Trailer和FileInfo。正如图中所示的，Trailer中有指针指向其他数据块的起始点。File Info中记录了文件的一些Meta信息，例如：AVG_KEY_LEN, AVG_VALUE_LEN, LAST_KEY, COMPARATOR, MAX_SEQ_ID_KEY等。Data Index和Meta Index块记录了每个Data块和Meta块的起始点。

Data Block是HBase I/O的基本单元，为了提高效率，HRegionServer中有基于LRU的Block Cache机制。每个Data块的大小可以在创建一个Table的时候通过参数指定，大号的Block有利于顺序Scan，小号Block利于随机查询。每个Data块除了开头的Magic以外就是一个个KeyValue对拼接而成, Magic内容就是一些随机数字，目的是防止数据损坏。后面会详细介绍每个KeyValue对的内部构造。

HFile里面的每个KeyValue对就是一个简单的byte数组。但是这个byte数组里面包含了很多项，并且有固定的结构。我们来看看里面的具体结构：

开始是两个固定长度的数值，分别表示Key的长度和Value的长度。紧接着是Key，开始是固定长度的数值，表示RowKey的长度，紧接着是RowKey，然后是固定长度的数值，表示Family的长度，然后是Family，接着是Qualifier，然后是两个固定长度的数值，表示Time Stamp和Key Type（Put/Delete）。Value部分没有这么复杂的结构，就是纯粹的二进制数据了。

HLogFile

上图中示意了HLog文件的结构，其实HLog文件就是一个普通的Hadoop Sequence File，Sequence File 的Key是HLogKey对象，HLogKey中记录了写入数据的归属信息，除了table和region名字外，同时还包括 sequence number和timestamp，timestamp是“写入时间”，sequence number的起始值为0，或者是最近一次存入文件系统中sequence number。

HLog Sequece File的Value是HBase的KeyValue对象，即对应HFile中的KeyValue，可参见上文描述。

结束

本文对HBase技术在功能和设计上进行了大致的介绍，由于篇幅有限，本文没有过多深入地描述HBase的一些细节技术。目前一淘的存储系统就是基于HBase技术搭建的，后续将介绍“一淘分布式存储系统”，通过实际案例来更多的介绍HBase应用。

Kafka是由LinkedIn开发的一个分布式的消息系统，使用Scala编写，它以可水平扩展和高吞吐率而被广泛使用。目前越来越多的开源分布式处理系统如Cloudera、Apache Storm、Spark都支持与Kafka集成。InfoQ一直在紧密关注Kafka的应用以及发展，“Kafka剖析”专栏将会从架构设计、实现、应用场景、性能等方面深度解析Kafka。

背景介绍

Kafka创建背景

Kafka是一个消息系统，原本开发自LinkedIn，用作LinkedIn的活动流（Activity Stream）和运营数据处理管道（Pipeline）的基础。现在它已被多家不同类型的公司 作为多种类型的数据管道和消息系统使用。

活动流数据是几乎所有站点在对其网站使用情况做报表时都要用到的数据中最常规的部分。活动数据包括页面访问量（Page View）、被查看内容方面的信息以及搜索情况等内容。这种数据通常的处理方式是先把各种活动以日志的形式写入某种文件，然后周期性地对这些文件进行统计分析。运营数据指的是服务器的性能数据（CPU、IO使用率、请求时间、服务日志等等数据)。运营数据的统计方法种类繁多。

近年来，活动和运营数据处理已经成为了网站软件产品特性中一个至关重要的组成部分，这就需要一套稍微更加复杂的基础设施对其提供支持。

Kafka简介

Kafka是一种分布式的，基于发布/订阅的消息系统。主要设计目标如下：

• 以时间复杂度为O(1)的方式提供消息持久化能力，即使对TB级以上数据也能保证常数时间复杂度的访问性能。
• 高吞吐率。即使在非常廉价的商用机器上也能做到单机支持每秒100K条以上消息的传输。
• 支持Kafka Server间的消息分区，及分布式消费，同时保证每个Partition内的消息顺序传输。
• 同时支持离线数据处理和实时数据处理。
• Scale out：支持在线水平扩展。

为何使用消息系统

• 解耦

  在项目启动之初来预测将来项目会碰到什么需求，是极其困难的。消息系统在处理过程中间插入了一个隐含的、基于数据的接口层，两边的处理过程都要实现这一接口。这允许你独立的扩展或修改两边的处理过程，只要确保它们遵守同样的接口约束。

• 冗余

  有些情况下，处理数据的过程会失败。除非数据被持久化，否则将造成丢失。消息队列把数据进行持久化直到它们已经被完全处理，通过这一方式规避了数据丢失风险。许多消息队列所采用的"插入-获取-删除"范式中，在把一个消息从队列中删除之前，需要你的处理系统明确的指出该消息已经被处理完毕，从而确保你的数据被安全的保存直到你使用完毕。

• 扩展性

  因为消息队列解耦了你的处理过程，所以增大消息入队和处理的频率是很容易的，只要另外增加处理过程即可。不需要改变代码、不需要调节参数。扩展就像调大电力按钮一样简单。

• 灵活性 & 峰值处理能力

  在访问量剧增的情况下，应用仍然需要继续发挥作用，但是这样的突发流量并不常见；如果为以能处理这类峰值访问为标准来投入资源随时待命无疑是巨大的浪费。使用消息队列能够使关键组件顶住突发的访问压力，而不会因为突发的超负荷的请求而完全崩溃。

• 可恢复性

  系统的一部分组件失效时，不会影响到整个系统。消息队列降低了进程间的耦合度，所以即使一个处理消息的进程挂掉，加入队列中的消息仍然可以在系统恢复后被处理。

• 顺序保证

  在大多使用场景下，数据处理的顺序都很重要。大部分消息队列本来就是排序的，并且能保证数据会按照特定的顺序来处理。Kafka保证一个Partition内的消息的有序性。

• 缓冲

  在任何重要的系统中，都会有需要不同的处理时间的元素。例如，加载一张图片比应用过滤器花费更少的时间。消息队列通过一个缓冲层来帮助任务最高效率的执行———写入队列的处理会尽可能的快速。该缓冲有助于控制和优化数据流经过系统的速度。

• 异步通信

  很多时候，用户不想也不需要立即处理消息。消息队列提供了异步处理机制，允许用户把一个消息放入队列，但并不立即处理它。想向队列中放入多少消息就放多少，然后在需要的时候再去处理它们。

常用Message Queue对比

• RabbitMQ

  RabbitMQ是使用Erlang编写的一个开源的消息队列，本身支持很多的协议：AMQP，XMPP, SMTP, STOMP，也正因如此，它非常重量级，更适合于企业级的开发。同时实现了Broker构架，这意味着消息在发送给客户端时先在中心队列排队。对路由，负载均衡或者数据持久化都有很好的支持。

• Redis

  Redis是一个基于Key-Value对的NoSQL数据库，开发维护很活跃。虽然它是一个Key-Value数据库存储系统，但它本身支持MQ功能，所以完全可以当做一个轻量级的队列服务来使用。对于RabbitMQ和Redis的入队和出队操作，各执行100万次，每10万次记录一次执行时间。测试数据分为128Bytes、512Bytes、1K和10K四个不同大小的数据。实验表明：入队时，当数据比较小时Redis的性能要高于RabbitMQ，而如果数据大小超过了10K，Redis则慢的无法忍受；出队时，无论数据大小，Redis都表现出非常好的性能，而RabbitMQ的出队性能则远低于Redis。

• ZeroMQ

  ZeroMQ号称最快的消息队列系统，尤其针对大吞吐量的需求场景。ZeroMQ能够实现RabbitMQ不擅长的高级/复杂的队列，但是开发人员需要自己组合多种技术框架，技术上的复杂度是对这MQ能够应用成功的挑战。ZeroMQ具有一个独特的非中间件的模式，你不需要安装和运行一个消息服务器或中间件，因为你的应用程序将扮演这个服务器角色。你只需要简单的引用ZeroMQ程序库，可以使用NuGet安装，然后你就可以愉快的在应用程序之间发送消息了。但是ZeroMQ仅提供非持久性的队列，也就是说如果宕机，数据将会丢失。其中，Twitter的Storm 0.9.0以前的版本中默认使用ZeroMQ作为数据流的传输（Storm从0.9版本开始同时支持ZeroMQ和Netty作为传输模块）。

• ActiveMQ

  ActiveMQ是Apache下的一个子项目。 类似于ZeroMQ，它能够以代理人和点对点的技术实现队列。同时类似于RabbitMQ，它少量代码就可以高效地实现高级应用场景。

• Kafka/Jafka

  Kafka是Apache下的一个子项目，是一个高性能跨语言分布式发布/订阅消息队列系统，而Jafka是在Kafka之上孵化而来的，即Kafka的一个升级版。具有以下特性：快速持久化，可以在O(1)的系统开销下进行消息持久化；高吞吐，在一台普通的服务器上既可以达到10W/s的吞吐速率；完全的分布式系统，Broker、Producer、Consumer都原生自动支持分布式，自动实现负载均衡；支持Hadoop数据并行加载，对于像Hadoop的一样的日志数据和离线分析系统，但又要求实时处理的限制，这是一个可行的解决方案。Kafka通过Hadoop的并行加载机制统一了在线和离线的消息处理。Apache Kafka相对于ActiveMQ是一个非常轻量级的消息系统，除了性能非常好之外，还是一个工作良好的分布式系统。

Kafka架构

Terminology

• Broker

  Kafka集群包含一个或多个服务器，这种服务器被称为broker

• Topic

  每条发布到Kafka集群的消息都有一个类别，这个类别被称为Topic。（物理上不同Topic的消息分开存储，逻辑上一个Topic的消息虽然保存于一个或多个broker上但用户只需指定消息的Topic即可生产或消费数据而不必关心数据存于何处）

• Partition

  Parition是物理上的概念，每个Topic包含一个或多个Partition.

• Producer

  负责发布消息到Kafka broker

• Consumer

  消息消费者，向Kafka broker读取消息的客户端。

• Consumer Group

  每个Consumer属于一个特定的Consumer Group（可为每个Consumer指定group name，若不指定group name则属于默认的group）。

Kafka拓扑结构

如上图所示，一个典型的Kafka集群中包含若干Producer（可以是web前端产生的Page View，或者是服务器日志，系统CPU、Memory等），若干broker（Kafka支持水平扩展，一般broker数量越多，集群吞吐率越高），若干Consumer Group，以及一个Zookeeper集群。Kafka通过Zookeeper管理集群配置，选举leader，以及在Consumer Group发生变化时进行rebalance。Producer使用push模式将消息发布到broker，Consumer使用pull模式从broker订阅并消费消息。

Topic & Partition

Topic在逻辑上可以被认为是一个queue，每条消费都必须指定它的Topic，可以简单理解为必须指明把这条消息放进哪个queue里。为了使得Kafka的吞吐率可以线性提高，物理上把Topic分成一个或多个Partition，每个Partition在物理上对应一个文件夹，该文件夹下存储这个Partition的所有消息和索引文件。若创建topic1和topic2两个topic，且分别有13个和19个分区，则整个集群上会相应会生成共32个文件夹（本文所用集群共8个节点，此处topic1和topic2 replication-factor均为1），如下图所示。

每个日志文件都是一个log entrie序列，每个log entrie包含一个4字节整型数值（值为N+5），1个字节的"magic value"，4个字节的CRC校验码，其后跟N个字节的消息体。每条消息都有一个当前Partition下唯一的64字节的offset，它指明了这条消息的起始位置。磁盘上存储的消息格式如下：

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

这个log entries并非由一个文件构成，而是分成多个segment，每个segment以该segment第一条消息的offset命名并以“.kafka”为后缀。另外会有一个索引文件，它标明了每个segment下包含的log entry的offset范围，如下图所示。

因为每条消息都被append到该Partition中，属于顺序写磁盘，因此效率非常高（经验证，顺序写磁盘效率比随机写内存还要高，这是Kafka高吞吐率的一个很重要的保证）。

对于传统的message queue而言，一般会删除已经被消费的消息，而Kafka集群会保留所有的消息，无论其被消费与否。当然，因为磁盘限制，不可能永久保留所有数据（实际上也没必要），因此Kafka提供两种策略删除旧数据。一是基于时间，二是基于Partition文件大小。例如可以通过配置$KAFKA_HOME/config/server.properties，让Kafka删除一周前的数据，也可在Partition文件超过1GB时删除旧数据，配置如下所示。

　　 # The minimum age of a log file to be eligible for deletion log.retention.hours=168 # The maximum size of a log segment file. When this size is reached a new log segment will be created. log.segment.bytes=1073741824 # The interval at which log segments are checked to see if they can be deleted according to the retention policies log.retention.check.interval.ms=300000 # If log.cleaner.enable=true is set the cleaner will be enabled and individual logs can then be marked for log compaction. log.cleaner.enable=false 

这里要注意，因为Kafka读取特定消息的时间复杂度为O(1)，即与文件大小无关，所以这里删除过期文件与提高Kafka性能无关。选择怎样的删除策略只与磁盘以及具体的需求有关。另外，Kafka会为每一个Consumer Group保留一些metadata信息——当前消费的消息的position，也即offset。这个offset由Consumer控制。正常情况下Consumer会在消费完一条消息后递增该offset。当然，Consumer也可将offset设成一个较小的值，重新消费一些消息。因为offet由Consumer控制，所以Kafka broker是无状态的，它不需要标记哪些消息被哪些消费过，也不需要通过broker去保证同一个Consumer Group只有一个Consumer能消费某一条消息，因此也就不需要锁机制，这也为Kafka的高吞吐率提供了有力保障。

Producer消息路由

Producer发送消息到broker时，会根据Paritition机制选择将其存储到哪一个Partition。如果Partition机制设置合理，所有消息可以均匀分布到不同的Partition里，这样就实现了负载均衡。如果一个Topic对应一个文件，那这个文件所在的机器I/O将会成为这个Topic的性能瓶颈，而有了Partition后，不同的消息可以并行写入不同broker的不同Partition里，极大的提高了吞吐率。可以在$KAFKA_HOME/config/server.properties中通过配置项num.partitions来指定新建Topic的默认Partition数量，也可在创建Topic时通过参数指定，同时也可以在Topic创建之后通过Kafka提供的工具修改。

在发送一条消息时，可以指定这条消息的key，Producer根据这个key和Partition机制来判断应该将这条消息发送到哪个Parition。Paritition机制可以通过指定Producer的paritition. class这一参数来指定，该class必须实现kafka.producer.Partitioner接口。本例中如果key可以被解析为整数则将对应的整数与Partition总数取余，该消息会被发送到该数对应的Partition。（每个Parition都会有个序号,序号从0开始）

import kafka.producer.Partitioner; import kafka.utils.VerifiableProperties;  public class JasonPartitioner<T> implements Partitioner {      public JasonPartitioner(VerifiableProperties verifiableProperties) {}      @Override     public int partition(Object key, int numPartitions) {         try {             int partitionNum = Integer.parseInt((String) key);             return Math.abs(Integer.parseInt((String) key) % numPartitions);         } catch (Exception e) {             return Math.abs(key.hashCode() % numPartitions);         }     } } 

如果将上例中的类作为partition.class，并通过如下代码发送20条消息（key分别为0，1，2，3）至topic3（包含4个Partition）。

public void sendMessage() throws InterruptedException{ 　　for(int i = 1; i <= 5; i++){ 　　      List messageList = new ArrayList<KeyedMessage<String, String>>(); 　　      for(int j = 0; j < 4; j++）{ 　　          messageList.add(new KeyedMessage<String, String>("topic2", j+"", "The " + i + " message for key " + j)); 　　      } 　　      producer.send(messageList);     } 　　producer.close(); } 

则key相同的消息会被发送并存储到同一个partition里，而且key的序号正好和Partition序号相同。（Partition序号从0开始，本例中的key也从0开始）。下图所示是通过Java程序调用Consumer后打印出的消息列表。

Consumer Group

（本节所有描述都是基于Consumer hight level API而非low level API）。

使用Consumer high level API时，同一Topic的一条消息只能被同一个Consumer Group内的一个Consumer消费，但多个Consumer Group可同时消费这一消息。

这是Kafka用来实现一个Topic消息的广播（发给所有的Consumer）和单播（发给某一个Consumer）的手段。一个Topic可以对应多个Consumer Group。如果需要实现广播，只要每个Consumer有一个独立的Group就可以了。要实现单播只要所有的Consumer在同一个Group里。用Consumer Group还可以将Consumer进行自由的分组而不需要多次发送消息到不同的Topic。

实际上，Kafka的设计理念之一就是同时提供离线处理和实时处理。根据这一特性，可以使用Storm这种实时流处理系统对消息进行实时在线处理，同时使用Hadoop这种批处理系统进行离线处理，还可以同时将数据实时备份到另一个数据中心，只需要保证这三个操作所使用的Consumer属于不同的Consumer Group即可。下图是Kafka在Linkedin的一种简化部署示意图。

下面这个例子更清晰地展示了Kafka Consumer Group的特性。首先创建一个Topic (名为topic1，包含3个Partition)，然后创建一个属于group1的Consumer实例，并创建三个属于group2的Consumer实例，最后通过Producer向topic1发送key分别为1，2，3的消息。结果发现属于group1的Consumer收到了所有的这三条消息，同时group2中的3个Consumer分别收到了key为1，2，3的消息。如下图所示。

Push vs. Pull

作为一个消息系统，Kafka遵循了传统的方式，选择由Producer向broker push消息并由Consumer从broker pull消息。一些logging-centric system，比如Facebook的Scribe和Cloudera的Flume，采用push模式。事实上，push模式和pull模式各有优劣。

push模式很难适应消费速率不同的消费者，因为消息发送速率是由broker决定的。push模式的目标是尽可能以最快速度传递消息，但是这样很容易造成Consumer来不及处理消息，典型的表现就是拒绝服务以及网络拥塞。而pull模式则可以根据Consumer的消费能力以适当的速率消费消息。

对于Kafka而言，pull模式更合适。pull模式可简化broker的设计，Consumer可自主控制消费消息的速率，同时Consumer可以自己控制消费方式——即可批量消费也可逐条消费，同时还能选择不同的提交方式从而实现不同的传输语义。

Kafka delivery guarantee

有这么几种可能的delivery guarantee：

• At most once 消息可能会丢，但绝不会重复传输
• At least one 消息绝不会丢，但可能会重复传输

• Exactly once 每条消息肯定会被传输一次且仅传输一次，很多时候这是用户所想要的。

  当Producer向broker发送消息时，一旦这条消息被commit，因数replication的存在，它就不会丢。但是如果Producer发送数据给broker后，遇到网络问题而造成通信中断，那Producer就无法判断该条消息是否已经commit。虽然Kafka无法确定网络故障期间发生了什么，但是Producer可以生成一种类似于主键的东西，发生故障时幂等性的重试多次，这样就做到了Exactly once。截止到目前(Kafka 0.8.2版本，2015-03-04)，这一Feature还并未实现，有希望在Kafka未来的版本中实现。（所以目前默认情况下一条消息从Producer到broker是确保了At least once，可通过设置Producer异步发送实现At most once）。

  接下来讨论的是消息从broker到Consumer的delivery guarantee语义。（仅针对Kafka consumer high level API）。Consumer在从broker读取消息后，可以选择commit，该操作会在Zookeeper中保存该Consumer在该Partition中读取的消息的offset。该Consumer下一次再读该Partition时会从下一条开始读取。如未commit，下一次读取的开始位置会跟上一次commit之后的开始位置相同。当然可以将Consumer设置为autocommit，即Consumer一旦读到数据立即自动commit。如果只讨论这一读取消息的过程，那Kafka是确保了Exactly once。但实际使用中应用程序并非在Consumer读取完数据就结束了，而是要进行进一步处理，而数据处理与commit的顺序在很大程度上决定了消息从broker和consumer的delivery guarantee semantic。

• 读完消息先commit再处理消息。这种模式下，如果Consumer在commit后还没来得及处理消息就crash了，下次重新开始工作后就无法读到刚刚已提交而未处理的消息，这就对应于At most once

• 读完消息先处理再commit。这种模式下，如果在处理完消息之后commit之前Consumer crash了，下次重新开始工作时还会处理刚刚未commit的消息，实际上该消息已经被处理过了。这就对应于At least once。在很多使用场景下，消息都有一个主键，所以消息的处理往往具有幂等性，即多次处理这一条消息跟只处理一次是等效的，那就可以认为是Exactly once。（笔者认为这种说法比较牵强，毕竟它不是Kafka本身提供的机制，主键本身也并不能完全保证操作的幂等性。而且实际上我们说delivery guarantee 语义是讨论被处理多少次，而非处理结果怎样，因为处理方式多种多样，我们不应该把处理过程的特性——如是否幂等性，当成Kafka本身的Feature）

• 如果一定要做到Exactly once，就需要协调offset和实际操作的输出。精典的做法是引入两阶段提交。如果能让offset和操作输入存在同一个地方，会更简洁和通用。这种方式可能更好，因为许多输出系统可能不支持两阶段提交。比如，Consumer拿到数据后可能把数据放到HDFS，如果把最新的offset和数据本身一起写到HDFS，那就可以保证数据的输出和offset的更新要么都完成，要么都不完成，间接实现Exactly once。（目前就high level API而言，offset是存于Zookeeper中的，无法存于HDFS，而low level API的offset是由自己去维护的，可以将之存于HDFS中）

总之，Kafka默认保证At least once，并且允许通过设置Producer异步提交来实现At most once。而Exactly once要求与外部存储系统协作，幸运的是Kafka提供的offset可以非常直接非常容易得使用这种方式。

鉴于自己的毕设是与视频检索有关，而在图像和视频检索领域中，D.Lowe和他的“亲儿子”——SIFT( Scale Invariant Feature Transform )算法是不能错过的经典论题，我在之前闲逛过的一个技术博客站点中找到了介绍这一经典算法基本概念的文章，原文地址：http://www.dakaren.com/index.php/archives/639.htm   

1、SIFT算法基本概念

Sift是David Lowe于1999年提出的局部特征描述子，可以处理两幅图像之间发生平移、旋转、仿射变换情况下的匹配问题，具有良好的不变性和很强的匹配能力。SIFT算法是一种提取局部特征的算法，也是一种模式识别技术，其基本思想是在尺度空间寻找极值点，提取位置，尺度，旋转不变量，它主要包括两个阶段，一个是Sift特征的生成，即从多幅图像中提取对尺度缩放、旋转、亮度变化无关的特征向量；第二阶段是Sift特征向量的匹配。Sift及其扩展算法已被证实在同类描述子中具有最强的健壮性，目前是国内外研究的热点。

2、SIFT算法的主要特点：

a) SIFT特征是图像的局部特征，其对旋转、尺度缩放、亮度变化保持不变 性，对视角变化、仿射变换、噪声也保持一定程度的稳定性，而对物体运动、遮 挡、噪声等因素也保持较好的可匹配性，从而可以实现差异较大的两幅图像之间 特征的匹配。

b) 独特性(Distinctiveness)好，信息量丰富，适用于在海量特征数据库中进行 快速、准确的匹配，比原有的harris点匹配方式具有更高的匹配准确度。

c) 多量性，即使少数的几个物体也可以产生大量SIFT特征向量。

d) 高速性，经优化的SIFT匹配算法甚至可以达到实时的要求。

e) 可扩展性，可以很方便的与其他形式的特征向量进行联合。

SIFT算法基于图像特征尺度选择的思想，建立图像的多尺度空间，在不同尺度下检测到同一个特征点，确定特征点位置的同时确定其所在尺度，以达到尺度抗缩放的目的。剔除一些对比度较低的点以及边缘响应点，并提取旋转不变特征描述符以达到抗仿射变换的目的。

3、SIFT算法步骤：

1） 构建尺度空间，检测极值点，获得尺度不变性；

2） 特征点过滤并进行精确定位；

3） 为每个关键点指定方向参数

4） 生成关键点的描述子

5） 当两幅图像的Sift特征向量生成以后，下一步就可以采用关键点特征向 量的欧式距离来作为两幅图像中关键点的相似性判定度量。取一幅图中的某个关键点，通过遍历找到另一幅图中的距离最近的两个关键点。在这两个关键点中，如果次近距离除以最近距离小于某个阙值，则判定为一对匹配点。降低这个比例阈值，SIFT匹配点数目会减少，但更加稳定。

4、SIFT算法发展历程：

Sift算子最早是由David.G.Lowe于1999年提出的，当时主要用于对象识别。2004年David.G.Lowe对该算子做了全面的总结及更深入的发展和完善，正式提出了一种基于尺度空间的、对图像缩放、旋转甚至仿射变换保持不变性的图像局部特征描述算子——Sift（ Scale Invariant Feature Transform )算子，即尺度不变特征变换。Rob Hess 基于GSL和Opencv编写了相应的C语言程序，后来Y.Ke将其描述子部分用PCA代替直方图的方式，对其进行改进。在Mikolajczyk对包括Sift算子在内的十种局部描述子所做的不变性对比实验中，Sift及其扩展算法已被证实在同类描述子中具有最强的健壮性。

主要文献：

1）David G. Lowe, “Object recognition from local scale-invariant features,” International Conference on Computer Vision, Corfu, Greece  2）David G. Lowe, “Distinctive image features from scale-invariant keypoints,” International Journal of Computer Vision,

3）Y. Ke and R. Sukthankar. PCA-SIFT: A More Distinctive Representation for Local Image Descriptors.Computer Vision and Pattern Recognition, 2004

5、关于局部不变特征

1)局部不变特征的概念

局部不变特征就是由局部邻域所构成的一个图像模式。局部不变特征可以是点集，也可以是边缘集合，或者一些小的图像块集合，甚至是上述集合的复合体。局部不变特征认为图像中总是存在一些特殊的区域，这些区域中的特征比其它图像区域的特征更加稳定，信息含量更高，能够表征图像的内容。局部不变特征的局部是指特征只是图像的局部区域，不变性是指该特征不会因为图像经历了各种变换而发生变化。

2）局部不变特征特点

局部不变特征的种类繁多，适合不同的特征提取场合，各自独立性较强，相互之间可以组合和借鉴。复合类型的局部不变特征可能会增加计算负担，但是能够取得更好的性能。

3）局部不变特征的应用

局部不变特征是一种十分有效的工具，大量研究表明它能够适应各种图像处理的应用场合，特别是在模拟人类视觉系统的物体识别领域，拥有强大的应用性。从直观的人类视觉印象来看，人类视觉对物体的描述也是局部化的，基于局部不变特征的图像识别方法十分接近于人类视觉机理，通过局部化的特征组合，形成对目标物体的整体印象，这就为局部不变特征提取方法提供了生物学上的解释，因此局部不变特征也得到了广泛应用。

4）特征描述符

特征描述符(Featrue Descriptors)指的是检测图像的局部特征（比如边缘、角点、轮廓等），然后根据匹配目标的需要进行特征的组合、变换，以形成易于匹配、稳定性好的特征向量，从而把图像匹配问题转化为特征的匹配问题，进而将特征的匹配问题转化为特征空间特征向量的聚类问题。

5）局部不变特征检测与局部不变特征区域的概念

局部不变特征检测就是从图像中检测出具有某种几何和光学不变性(geometric and photometric invariant)的局部不变特征区域。局部不变特征区域是以特征点(feature point or key point)为中心带有尺度信息的图像局部区域。局部不变特征认为，在大多数的图像中总能找到一些性质特殊的区域，它们可以稳定的提取，并且对各种图像变化具有良好的鲁棒性且包涵更多的图像内容信息。

6、基于局部不变特征的图像处理理论和技术主要包括四个部分：

1）图像尺度空间：图像数据包含大量混杂在一起的特征信息，按照局部不变特征的思想，这些特征信息是归属于不同类型不同属性的目标物体，其位置和控制区域各不相同，不同分布和参数的特征相互叠加和组合，这使得特征提取变得困难，所以需将这些特征进行一定的“分离”操作，将各类特征分散到整个图像数据空间中去，图像尺度空间就是为图像的各种不同类特征分离提供的一种数据表示法。

2）局部不变特征检测：在尺度空间内构造不变特征检测函数，生成对应尺度下的局部不变特征空间，检测其中具有一定特征显著性的局部不变特征区域，并把它作为特征描述的目标区域，确定每个特征的尺度系数，局部不变特征结构的位置和尺度范围。这些局部区域及其包含的信息形成对图像语义结构信息的表示，为进一步的特征描述提供图像内容的结构和范围信息。

3）局部不变特征描述：局部不变特征检测获得的特征仅仅给出了图像内容的结构信息，局部不变特征区域还需要从图像尺度空间表示的数据形式转化特征描述向量。局部不变特征描述就是用局部不变特征描述符(Local feature descriptor)去描述局部不变特征区域，用尽可能相互独立和完备的特征描述数据来表示复杂组合的目标物体，完整详细地描述图像内容，给出图像的语义信息。

4）特征匹配和检索：特征提取的最终目的是使用这些特征来进行目标识别和特征的检索，通过对特征描述空间中的特征数据进行分类、匹配和检索，实现各种图像识别应用。由于其良好的鲁棒性和抗干扰性，使的它作为目标识别中机器学习样本描述的首选特征，图像和视频检索方法也大都采用局部不变特征作为学习和检索的依据。

7、局部不变特征发展方向

目前，局部不变特征主要分为两个发展方向：

1) 结构化的局部不变特征提取模型，也就是特征提取模型分为四个较为清晰的处理模块（上文中有提到）。在局部不变特征检测方面，D.Lowe提出基于扩散方程的尺度不变的SIFT特征检测方法，以及由角点检测发展而来的Multi-scale Harris检测，具有仿射不变性的Harris-Laplace/Affine检测等，目前局部不变特征检测方法逐渐向着检测具有多种不变性和抗干扰性强的局部不变特征的方向发展。局部不变特征描述技术更加广泛，其中以SIFT，GLOH,Steerable Filters,Shape Context,Complex Filters等为主要特征描述符。特征检索和匹配模块一般是面向图像模式识别的具体应用场合，如图像检索，机器学习的样本特征集合，目标识别中的样本特征数据库等，同时在视频的检索领域也获得不错的效果。

2) 模仿人类的视觉系统，通过模仿人类视觉系统的运作原理提出了显著性区域理论。这一理论认为图像中的每个局部区域的重要性和影响范围并非同等重要，即特征不是同等显著的，其主要理论来源是Marr的计算机视觉理论和Treisman的特征整合理论，一般也称为“原子论”。该理论认为视觉的过程开始于对物体的特征性质和简单组成部分的分析，是从局部性

质到大范围性质，图像中的每个局部不变特征的视觉显著性是不同的，所以在局部不变特征的提取和描述时也遵循与人眼视觉注意选择原理相类似的机制（Visual Selective Attention Mechanism）。

8、软件：SIFT Keypoint Detector

该软件是可以在Linux或Windows系统中运行的汇编代码形式的SIFT特征点检测器， 它可以输出特征点和可以匹配到一个简单的ASCII格式文件需要的所有信息。 所提供的MATLAB程序和示例C代码可以读取特征点并根据它们对两幅图片进行匹配。

9、应用前景

SIFT算法是模式识别的一种高效手段，凡模式识别的应用方面都可以运用SIFT算法来改进识别速度。

医学：运动学人体机能研究

仿生学：人工模拟生物

人工智能：智能机器人、智能驾驶

刑侦技术：跟踪

军事用途：敌友识别（战机、战舰、潜艇、雷达跟踪等等）

1、我先要给大家讲一个概念：spring 的多数据源事务，这是民间的说法。官方的说法是：spring 的分布式事务。明白了这个概念，问题就好解决了。

2、分布式事务的应用场景：工程中使用两个及以上数据库中，就要考虑使用分布式事务管理，否则事务不能回滚。

3、现有两种开源的第三方jar支持spring的分布式事务管理，它们分别是：jotm和Atomikos。通过google可以找到下载的链接，其中atomikos的下载需要先填写email信息，再登录email找到链接去下载。下载地址分别：

1、http://jotm.objectweb.org/

2、http://www.atomikos.com/Main/InstallingTransactionsEssentials

我使用的是jotm。

4、基于spring+ibatis的环境下配置jotm的方法很简单。只城要修改spring数据源的配置及事务的配置及可。以下是我的配置，供参考。

文件放在classpath下面，也就是src下面。

6、需要jtom的以下jar：

7、经过测试，配置是成功的。事务可以回滚。