数据高效处理的秘诀——Kudu实战

背景

使用Kudu之前，我们的Online Report采用都HDFS/Parquet on Impala的架构，数据每隔一小时通过MapReduce从生产db增量同步到HDFS，再通过HIVE/MAPREDUCE增量MERGE到数仓中，最终通过IMPALA查询在报表展示。这种架构可以支持超大数据集的查询和分析，但是数据更新执行过程很久，效率较差。而Apache Kudu结合了Hbase和HDFS的优势，可以同时提供高效的随机访问以及数据扫描能力，支持数据的实时插入和分析，为Online Report提供了另外一种选择。

Kudu简介

Kudu设计优势

• OLAP工作流的快速处理能够快速更新/获取新数据，并进行在线查询分析
• 列式存储结构
• 集成Hadoop生态组件Spark/Flume/MapReduce/Impala等
• 服务的高可用性TabletServer以及Master都使用*Raft*一致性算法。该算法确保只要可用副本数为总数的一半以上，Tablet就可以进行读写操作。即使在leader发生故障的情况下，follower也可以为程序提供只读服务。
• 结构化数据模型类关系型数据库，强字段类型，主键索引，表分区设计
• NOSQL APIs支持 C++, Java and Python (INSERT/UPDATE/SCAN/DELETE等)

Kudu整体架构

下图展示了一个三个Master和N个TabletServer的Kudu集群

角色概念：

• Tablet一个table会由多个连续的tablet组成，这里的tablet类似于其他数据存储引擎或者传统数据库中的一个分区。为了容灾和高可用性，每一个tablet会备份至 N（备份数为奇数，最多为7个）个tablet server上，并且其中一个被选举为leader。任何副本都可以进行读操作，写操作时需要在为tablet服务的一组tablet server之间达成一致性
• Master1. 为了保证高可用性，master建议由3个节点部署，其中一个会是leader，如果当前的leader消失，则通过*Raft Consensus Algorithm*选举出一个新的master。2. master会跟踪tablets、tablet servers、catalog table以及其他原数据的状态和信息，并协调client对metadata的操作3. 所有master的数据都存在一个tablet里，可以被复制到其他候选的master节点上
• Catalog Table  1. tables2. 存储表结构，位置以及状态3. tablet servers4. 存储管理的tablet列表，每个tablet所在的tablet server，tablet的当前状态，开始以及结束的key
• Logical ReplicationKudu副本复制操作与HDFS不同，HDFS需要完全通过网络传输来满足他所需要的副本数，而Kudu除了INSERT/UPDATE操作，其他操作都在节点本地执行，如DELETE操作，会将DELETE操作发送至每个tablet server，在他们等本地执行删除操作，极大的降低了集群的网络负载，另外，物理操作COMPACTION，也是本地单独执行，并且不需要在同一时间执行所有副本的压缩，这样能够减少由于COMPACTION以及大量写入操作，导致的高负载。

Kudu实战

表结构设计  

创建一个Kudu表

CREATE TABLE ods.info (   

userid INT NOT NULL,

username STRING NULL,

department STRING NULL ENCODING AUTO_ENCODING COMPRESSION DEFAULT_COMPRESSION, –字段类型、编码、以及压缩

email STRING NULL,  

PRIMARY KEY (userid) –主键

)

PARTITION BY HASH (userid) PARTITIONS 20–分区方式

STORED AS KUDU

TBLPROPERTIES(‘kudu.master_addresses’=KUDU_MASTER_ADDRESS) –指定kudu master地址

• 表结构主要有三块内容：字段设计1. 字段数量不允许超过300个2. 除主键外，其他字段可以为空3. 每一个字段均可以设置自己的编码以及压缩方式4. Kudu1.7.0版本开始，支持Decimal字段类型，适用于金融和特定的算数运算主键设计1. 建表必须包含主键，主键字段必须列在Schema的最前端2. 建表后，主键无法更改，只能重建表3. 不支持自增列4. 主键不能为空，并且不能为boolean、float或者double类型5. 主键的值无法被更新，但是可以被DELETE后，re-INSERT6. 主键即索引，tablet中的所有行都按照主键排序。查询时，对主键指定相等或范围的谓词，Kudu扫描表的时候会过滤掉不满足条件的行。分区设计1. 不允许您更改创建后如何分区表，但可以添加或删除范围分区2. 分区方式：哈希分区、范围分区以及多级分区。

• 表结构可以修改的内容：1. 表名可以重命名2. 主键名可以重命名3. 非主键字段可以被重命名、删除或新增4. 可以新增或删除范围分区
  
• 表结构最佳实践1. 根据自身业务场景，选择合适的分区方式，让读与写操作在所有tablet server上均匀分布。2. 根据应用查询的语句，设计合理的主键以及分区，保证读取数据时扫描最小的数据集3. 分区数量的设置，根据官方文档，每个分区的大小尽量控制在4G左右（单个tablet server最大存储8T/管理的tablets数量最大2000个≈4G），如果你的表数据量未来估算会在40G左右，那么你的分区数量可以设置10个。

JAVA API 的使用

KuduKuduClient client = (newKuduClientBuilder(masterAddress)).build();

KuduTable table =client.openTable(tableName);

KuduSession session = client.newSession();

session.setFlushMode(FlushMode.AUTO_FLUSH_BACKGROUND);

Insert insert = table.newInsert();

insert.getRow().addInt(“userid”, 100001);

session.apply(insert);

Kudu实战总结

使用情况

• 目前通过canal消费生产Mysql-Binlog到Kafka，再通过Spring-Kafka-Consumer实时写入Kudu底层表
• 有一部分SqlServer的生产数据，通过KuduMapReduce的方式，每半个小时，批量更新到Kudu底层表
• Online Report通过Kudu on Impala的方式，查询结果并展示

坑点

• 时钟服务NTP配置不合理，会导致Kudu服务直接崩溃，建议根据官方的推荐来配置NTP，另外可以通过修改参数max_clock_sync_error_usec值，来提高Kudu对时间偏差的容忍程度
• 在Impala中对Kudu表进行alter table A rename to B，只会更改impala的元数据，而不会更改任何Kudu的元数据，可以通过，先修改Impala元数据alter table A rename to B 后，再修改Kudu元数据alter table A set TBLPROPERTIES（’kudu.table_name’=’B’）
• 没有Rebalance功能，需要手工来做balance
• 在Kudu1.6.0之前，如果tablet server的某个磁盘坏了，那么整个tablet server就要重新format了，如果你的集群版本大于等于1.6.0并且损坏的盘并非WAL/Meta盘，那么你可以通过kusu fs update_dirs的方式来更新数据盘信息后，正常启动tablet server