一、背景介绍
在KUDU之前，大数据主要以两种方式存储；
（1）静态数据：
以 HDFS 引擎作为存储引擎，适用于高吞吐量的离线大数据分析场景。这类存储的局限性是数据无法进行随机的读写。
（2）动态数据：
以 HBase、Cassandra 作为存储引擎，适用于大数据随机读写场景。这类存储的局限性是批量读取吞吐量远不如 HDFS，不适用于批量数据分析的场景。
从上面分析可知，这两种数据在存储方式上完全不同，进而导致使用场景完全不同，但在真实的场景中，边界可能没有那么清晰，面对既需要随机读写，又需要批量分析的大数据场景，该如何选择呢？这个场景中，单种存储引擎无法满足业务需求，我们需要通过多种大数据工具组合来满足这一需求。
数据实时写入 HBase，实时的数据更新也在 HBase 完成，为了应对 OLAP 需求，我们定时（通常是 T+1 或者 T+H）将 HBase 数据写成静态的文件（如：Parquet）导入到 OLAP 引擎（如：HDFS）。这一架构能满足既需要随机读写，又可以支持 OLAP 分析的场景。但是缺点也比较明显：
(1)架构复杂。从架构上看，数据在HBase、消息队列、HDFS 间流转，涉及环节太多，运维成本很高。并且每个环节需要保证高可用，都需要维护多个副本，存储空间也有一定的浪费。最后数据在多个系统上，对数据安全策略、监控等都提出了挑战。
(2)时效性低。数据从HBase导出成静态文件是周期性的，一般这个周期是一天（或一小时），在时效性上不是很高。
(3)难以应对后续的更新。真实场景中，总会有数据是延迟到达的。如果这些数据之前已经从HBase导出到HDFS，新到的变更数据就难以处理了，一个方案是把原有数据应用上新的变更后重写一遍，但这代价又很高。
为了解决上述架构的这些问题，KUDU应运而生。KUDU的定位是Fast Analytics on Fast Data，是一个既支持随机读写、又支持 OLAP 分析的大数据存储引擎。
二、kudu基础
2.1 使用场景
适用于那些既有随机访问，也有批量数据扫描的复合场景、高计算量的场景、使用了高性能的存储设备，包括使用更多的内存、支持数据更新，避免数据反复迁移、支持跨地域的实时数据备份和查询。
2.2 kudu架构
与HDFS和HBase相似，Kudu使用单个的Master节点，用来管理集群的元数据，并且使用任意数量的Tablet Server（可对比理解HBase中的RegionServer角色）节点用来存储实际数据。可以部署多个Master节点来提高容错性。一个table表的数据，被分割成1个或多个Tablet，Tablet被部署在Tablet Server来提供数据读写服务。
一些基本概念：
Master：集群中的老大，负责集群管理、元数据管理等功能
Tablet Server: 集群中的小弟，负责数据存储，并提供数据读写服务。一个 tablet server 存储了table表的tablet 和为 tablet 向 client 提供服务。对于给定的 tablet，一个tablet server 充当 leader，其他 tablet server 充当该 tablet 的 follower 副本。只有 leader服务写请求，然而 leader 或 followers 为每个服务提供读请求 。一个 tablet server 可以服务多个 tablets ，并且一个 tablet 可以被多个 tablet servers 服务着。
Table(表)：一张table是数据存储在Kudu的tablet server中。表具有 schema 和全局有序的primary key（主键）。table 被分成称为 tablets 的 segments。
Tablet：一个 tablet 是一张 table连续的segment，tablet是kudu表的水平分区，类似于google Bigtable的tablet，或者HBase的region。每个tablet存储着一定连续range的数据（key），且tablet两两间的range不会重叠。一张表的所有tablet包含了这张表的所有key空间。与其它数据存储引擎或关系型数据库中的 partition（分区）相似。给定的tablet 冗余到多个 tablet 服务器上，并且在任何给定的时间点，其中一个副本被认为是leader tablet。任何副本都可以对读取进行服务，并且写入时需要在为 tablet 服务的一组 tablet server之间达成一致性。
三、kudu分区
为了提供可扩展性，Kudu 表被划分为称为 tablets 的单元，并分布在许多 tablet servers 上。行总是属于单个tablet 。将行分配给 tablet 的方法由在表创建期间设置的表的分区决定。 kudu提供了3种分区方式。
3.1 Range Partitioning ( 范围分区 )
范围分区可以根据存入数据的数据量，均衡的存储到各个机器上，防止机器出现负载不均衡现象.
1 /** 2 3 * 测试分区： 4 5 * RangePartition 6 7 */ 8 9 @Test 10 11 public void testRangePartition() throws KuduException { 12 13 //设置表的schema 14 15 LinkedList<ColumnSchema> columnSchemas = new LinkedList<ColumnSchema>(); 16 17 columnSchemas.add(newColumn("CompanyId", Type.INT32,true)); 18 19 columnSchemas.add(newColumn("WorkId", Type.INT32,false)); 20 21 columnSchemas.add(newColumn("Name", Type.STRING,false)); 22 23 columnSchemas.add(newColumn("Gender", Type.STRING,false)); 24 25 columnSchemas.add(newColumn("Photo", Type.STRING,false)); 26 27 28 29 //创建schema 30 31 Schema schema = new Schema(columnSchemas); 32 33 34 35 //创建表时提供的所有选项 36 37 CreateTableOptions tableOptions = new CreateTableOptions(); 38 39 //设置副本数 40 41 tableOptions.setNumReplicas(1); 42 43 //设置范围分区的规则 44 45 LinkedList<String> parcols = new LinkedList<String>(); 46 47 parcols.add("CompanyId"); 48 49 //设置按照那个字段进行range分区 50 51 tableOptions.setRangePartitionColumns(parcols); 52 53 54 55 /** 56 57 * range 58 59 * 0 < value < 10 60 61 * 10 <= value < 20 62 63 * 20 <= value < 30 64 65 * ........ 66 67 * 80 <= value < 90 68 69 * */ 70 71 int count=0; 72 73 for(int i =0;i<10;i++){ 74 75 //范围开始 76 77 PartialRow lower = schema.newPartialRow(); 78 79 lower.addInt("CompanyId",count); 80 81 82 83 //范围结束 84 85 PartialRow upper = schema.newPartialRow(); 86 87 count +=10; 88 89 upper.addInt("CompanyId",count); 90 91 92 93 //设置每一个分区的范围 94 95 tableOptions.addRangePartition(lower,upper); 96 97 } 98 99 100 101 try { 102 103 kuduClient.createTable("student",schema,tableOptions); 104 105 } catch (KuduException e) { 106 107 e.printStackTrace(); 108 109 } 110 111 kuduClient.close(); 112 113 }
3.2 Hash Partitioning (哈希分区)
哈希分区通过哈希值将行分配到许多 buckets ( 存储桶 )之一； 哈希分区是一种有效的策略，当不需要对表进行有序访问时。哈希分区对于在 tablet 之间随机散布这些功能是有效的，这有助于减轻热点和 tablet 大小不均匀。
1 /** 2 3 * 测试分区： 4 5 * hash分区 6 7 */ 8 9 @Test 10 11 public void testHashPartition() throws KuduException { 12 13 //设置表的schema 14 15 LinkedList<ColumnSchema> columnSchemas = new LinkedList<ColumnSchema>(); 16 17 columnSchemas.add(newColumn("CompanyId", Type.INT32,true)); 18 19 columnSchemas.add(newColumn("WorkId", Type.INT32,false)); 20 21 columnSchemas.add(newColumn("Name", Type.STRING,false)); 22 23 columnSchemas.add(newColumn("Gender", Type.STRING,false)); 24 25 columnSchemas.add(newColumn("Photo", Type.STRING,false)); 26 27 28 29 //创建schema 30 31 Schema schema = new Schema(columnSchemas); 32 33 34 35 //创建表时提供的所有选项 36 37 CreateTableOptions tableOptions = new CreateTableOptions(); 38 39 //设置副本数 40 41 tableOptions.setNumReplicas(1); 42 43 //设置范围分区的规则 44 45 LinkedList<String> parcols = new LinkedList<String>(); 46 47 parcols.add("CompanyId"); 48 49 //设置按照那个字段进行range分区 50 51 tableOptions.addHashPartitions(parcols,6); 52 53 try { 54 55 kuduClient.createTable("dog",schema,tableOptions); 56 57 } catch (KuduException e) { 58 59 e.printStackTrace(); 60 61 } 62 63 64 kuduClient.close(); 65 66 }
3.3 Multilevel Partitioning ( 多级分区 )
Kudu 允许一个表在单个表上组合多级分区。 当正确使用时，多级分区可以保留各个分区类型的优点，同时减少每个分区的缺点需求.
1 /** 2 3 * 测试分区： 4 5 * 多级分区 6 7 * Multilevel Partition 8 9 * 混合使用hash分区和range分区 10 11 * 12 13 * 哈希分区有利于提高写入数据的吞吐量，而范围分区可以避免tablet无限增长问题， 14 15 * hash分区和range分区结合，可以极大的提升kudu的性能 16 17 */ 18 19 @Test 20 21 public void testMultilevelPartition() throws KuduException { 22 23 //设置表的schema 24 25 LinkedList<ColumnSchema> columnSchemas = new LinkedList<ColumnSchema>(); 26 27 columnSchemas.add(newColumn("CompanyId", Type.INT32,true)); 28 29 columnSchemas.add(newColumn("WorkId", Type.INT32,false)); 30 31 columnSchemas.add(newColumn("Name", Type.STRING,false)); 32 33 columnSchemas.add(newColumn("Gender", Type.STRING,false)); 34 35 columnSchemas.add(newColumn("Photo", Type.STRING,false)); 36 37 38 39 //创建schema 40 41 Schema schema = new Schema(columnSchemas); 42 43 //创建表时提供的所有选项 44 45 CreateTableOptions tableOptions = new CreateTableOptions(); 46 47 //设置副本数 48 49 tableOptions.setNumReplicas(1); 50 51 //设置范围分区的规则 52 53 LinkedList<String> parcols = new LinkedList<String>(); 54 55 parcols.add("CompanyId"); 56 57 58 59 //hash分区 60 tableOptions.addHashPartitions(parcols,5); 61 62 63 64 //range分区 65 66 int count=0; 67 for(int i=0;i<10;i++){ 68 PartialRow lower = schema.newPartialRow(); 69 lower.addInt("CompanyId",count); 70 count+=10; 71 72 PartialRow upper = schema.newPartialRow(); 73 upper.addInt("CompanyId",count); 74 tableOptions.addRangePartition(lower,upper); 75 76 } 77 78 try { 79 80 kuduClient.createTable("cat",schema,tableOptions); 81 82 } catch (KuduException e) { 83 84 e.printStackTrace(); 85 86 } 87 88 kuduClient.close(); 89 90 }
四、kudu原理
4.1 表与schema
Kudu设计是面向结构化存储的，因此，Kudu的表需要用户在建表时定义它的Schema信息，这些Schema信息包含：列定义（含类型），Primary Key定义（用户指定的若干个列的有序组合）。数据的唯一性，依赖于用户所提供的Primary Key中的Column组合的值的唯一性。 Kudu提供了Alter命令来增删列，但位于Primary Key中的列是不允许删除的。 Kudu当前并不支持二级索引。 从用户角度来看，Kudu是一种存储结构化数据表的存储系统。在一个Kudu集群中可以定义任意数量的table，每个table都需要预先定义好schema。每个table的列数是确定的，每一列都需要有名字和类型，每个表中可以把其中一列或多列定义为主键。这么看来，Kudu更像关系型数据库，而不是像HBase、Cassandra和MongoDB这些NoSQL数据库。不过Kudu目前还不能像关系型数据一样支持二级索引。
Kudu使用确定的列类型，而不是类似于NoSQL的"everything is byte"。这可以带来两点好处： 确定的列类型使Kudu可以进行类型特有的编码。 可以提供 SQL-like 元数据给其他上层查询工具，比如BI工具。
4.2 kudu的底层数据模型
kudu的底层数据文件的存储，未采用HDFS这样的较高抽象层次的分布式文件系统，而是自行开发了一套可基于
Table/Tablet/Replica视图级别的底层存储系统。
这套实现基于如下的几个设计目标：
- 可提供快速的列式查询
- 可支持快速的随机更新
- 可提供更为稳定的查询性能保障
一张表会分成若干个tablet，每个tablet包括MetaData元信息及若干个RowSet，RowSet包含一个MemRowSet及若干个DiskRowSet，DiskRowSet中包含一个BloomFile、Ad_hoc Index、BaseData、DeltaMem及若干个RedoFile和UndoFile（UndoFile一般情况下只有一个）。
- MemRowSet：用于新数据insert及已在MemRowSet中的数据的更新，一个MemRowSet写满后会将数据刷到磁盘形成若干个DiskRowSet。每次到达32M生成一个DiskRowSet。
- DiskRowSet：用于老数据的变更（mutation），后台定期对DiskRowSet做compaction，以删除没用的数据及合并历史数据，减少查询过程中的IO开销。
- BloomFile：根据一个DiskRowSet中的key生成一个bloom filter，用于快速模糊定位某个key是否在DiskRowSet中存在。
- Ad_hocIndex：是主键的索引，用于定位key在DiskRowSet中的具体哪个偏移位置。
- BaseData是MemRowSet flush下来的数据，按列存储，按主键有序。
- UndoFile是基于BaseData之前时间的历史数据，通过在BaseData上apply UndoFile中的记录，可以获得历史数据。
- RedoFile是基于BaseData之后时间的变更（mutation）记录，通过在BaseData上apply RedoFile中的记录，可获得较新的数据。
- DeltaMem用于DiskRowSet中数据的变更mutation，先写到内存中，写满后flush到磁盘形成RedoFile。
MemRowSets可以对比理解成HBase中的MemStore, 而DiskRowSets可理解成HBase中的HFile。MemRowSets中的数据按照行试图进行存储，数据结构为B-Tree。
MemRowSets中的数据被Flush到磁盘之后，形成DiskRowSets。
DisRowSets中的数据，按照32MB大小为单位，按序划分为一个个的DiskRowSet。 DiskRowSet中的数据按照Column进行组织，与Parquet类似。
这是Kudu可支持一些分析性查询的基础。每一个Column的数据被存储在一个相邻的数据区域，而这个数据区域进一步被细分成一个个的小的Page单元，与HBase File中的Block类似，对每一个Column Page可采用一些Encoding算法，以及一些通用的Compression算法。 既然可对Column Page可采用Encoding以及Compression算法，那么，对单条记录的更改就会比较困难了。
前面提到了Kudu可支持单条记录级别的更新/删除，是如何做到的？
与HBase类似，也是通过增加一条新的记录来描述这次更新/删除操作的。DiskRowSet是不可修改了，那么 KUDU 要如何应对数据的更新呢？在KUDU中，把DiskRowSet分为了两部分：base data、delta stores。base data 负责存储基础数据，delta stores负责存储 base data 中的变更数据.
如上图所示，数据从 MemRowSet 刷到磁盘后就形成了一份 DiskRowSet（只包含 base data），每份 DiskRowSet 在内存中都会有一个对应的DeltaMemStore，负责记录此 DiskRowSet 后续的数据变更（更新、删除）。DeltaMemStore 内部维护一个 B-树索引，映射到每个 row_offset 对应的数据变更。DeltaMemStore 数据增长到一定程度后转化成二进制文件存储到磁盘，形成一个 DeltaFile，随着 base data 对应数据的不断变更，DeltaFile 逐渐增长。
4.3 kudu的写流程
如上图，当 Client 请求写数据时，先根据主键从Master Server中获取要访问的目标 Tablets，然后到依次对应的Tablet获取数据。
因为KUDU表存在主键约束，所以需要进行主键是否已经存在的判断，这里就涉及到之前说的索引结构对读写的优化了。一个Tablet中存在很多个RowSets，为了提升性能，我们要尽可能地减少要扫描的RowSets数量。
首先，我们先通过每个 RowSet 中记录的主键的（最大最小）范围，过滤掉一批不存在目标主键的RowSets，然后在根据RowSet中的布隆过滤器，过滤掉确定不存在目标主键的 RowSets，最后再通过RowSets中的 B-树索引，精确定位目标主键是否存在。
如果主键已经存在，则报错（主键重复），否则就进行写数据（写 MemRowSet）。
4.4 kudu的读流程
如上图，数据读取过程大致如下：先根据要扫描数据的主键范围，定位到目标的Tablets，然后读取Tablets 中的RowSets。
在读取每个RowSet时，先根据主键过滤要scan范围，然后加载范围内的base data，再找到对应的delta stores，应用所有变更，最后union上MemRowSet中的内容，返回数据给Client。
4.5 kudu的更新流程
数据更新的核心是定位到待更新数据的位置，这块与写入的时候类似，就不展开了，等定位到具体位置后，然后将变更写到对应的delta store 中。