DTCC2017-Kudu介绍-小米张震-final

Size: px

Start display at page:

Download "DTCC2017-Kudu介绍-小米张震-final"

源都
4 years ago
Views:

1 Kudu 介绍及实践小米张震

2 纲 Kudu 介绍 Kudu 与 Spark 的整合小米实践

3 存储系统的现状实时更新 HBase OLTP 随机读写数据更新的频次频繁更新追加只读 HDFS OLAP 批量扫描归档离线查询较频繁的查询查询的频次实时访问

4 Kudu 的设计目标实时更新 HBase OLTP 随机读写数据更新的频次频繁更新追加只读 HDFS OLAP 批量扫描 Kudu Fast Analytics on Fast Data 归档离线查询较频繁的查询查询的频次实时访问

5 Kudu 的设计目标性能 : 快速的批量扫描 2x Parquet 低延迟的随机读写 1ms on SSD 可扩展 : 400 nodes,1000s of nodes low MTTR 关系型数据模型 : 强 schema, 有限列, 不支持 BLOBs NoSQL APIs (insert/update/delete/scan), Java/C++ Client 事务支持 : 单的 ACID 支持与 Hadoop 态的集成 : Flume/Impala/Spark

6 Kudu 的数据模型有限固定列强类型 CREATE TABLE sales_by_year( year INT, sale_id INT COLPROPERTIES (encoding= bitshuffle ), amount INT, PRIMARY KEY (year, sale_id) ) PARTITION BY HASH (sale_id) PARTITIONS 4, RANGE (year) ( PARTITION 2014 <= VALUES <= 2016, PARTITION VALUE = 2017 ) STORED AS KUDU TBLPROPERTIES (replication=3); 列存储每列均可以设置 encoding 及压缩式主键索引范围分区和哈希分区多副本

7 分区和 Tablet Range (year) Hash (id) PARTITION BY HASH (id) PARTITIONS 4, RANGE (year) ( PARTITION VALUE = 2014 PARTITION VALUE = 2015 PARTITION VALUE = 2016 )

8 多备份 Range (year) Tablet Server Tablet 10 Tablet 3 Tablet Server Tablet 10 Tablet 7 Hash (id) Tablet Server Tablet 3 Tablet 2 Tablet Server Tablet10 Tablet 7

9 多备份 & raft 协议 Client 1. Client 发送写请求 6. 回复 Client 写成功 Tablet Server A Tablet 10 Leader 2. 写 WAL WAL 5. Leader 得知多数已经写成功 2. Leader 要求 Follower 进备份 Tablet Server A 3. 写 WAL 4. Follower 备份成功后返回状态给 Leader Tablet 10 Follower WAL Tablet Server A Tablet 10 Follower 3. 写 WAL WAL

10 多备份 & 容错多备份 : 强致, 低延迟允许从 Follower 进 Snapshot 读临时失败 (5 分钟内 ): Leader 临时失败 - 自动选主, 概在 5s 内完成 Follower 临时失败 - 少数的 Follower 失败不会影响读写永久失败 ( 超过 5 分钟 ): 踢出 Raft Config,Master 会重新选择台 TabletServer 部署新的 Replica

11 整体架构

12 Master 特殊的元表 : 表结构为 :type int8, id int64, meta string type 为 table 或 tablet id 为 table id 或 tablet id meta 为 protobuf 序列化后的元数据 table 元数据 schema,partition schema, 包含哪些 tablets tablet 元数据 replica locations tablet state Coordinator: 接收 TabletServer 的跳, 跟踪集群状态 create/alter/drop table 对 under-replicated 的 tablet 重新分配 replica

13 Tablet1 Tablet2 Tablet3 insert msg= foo Client timestamp Master 我想插数据 msg= foo, 应该发请求给谁? msg= foo 应该插到 Tablet1, 分布在 WXY, 另外你可能对 Tablet2 和 Tablet3 也感兴趣

14 Tablet 类 LSM(HBase): 每个 Tablet 包含个 MemRowSet 和多个 DiskRowSet; 新插的数据存储在 MemRowSet 中, 定期 flush 成 DiskRowSet 不同于 LSM, 每个 Row 只存在于个 RowSet 中 RowSet RowSet 内有序,RowSet 间序, 不同 RowSet 所包含的 key range 允许重叠 MemRowSet 内部使用 B 树存储,DiskRowSet 使用列存储 Tablet Server 内存 Tablet MemRowSet 磁盘 DiskRowSet [1, 2,, 10] DiskRowSet [7, 8,, 15] DiskRowSet [13, 14,, 20] DiskRowSet [18, 24,, 30] DiskRowSet [25, 29,, 40]

15 MVCC MVCC 对每都保存多个版本 MVCC 所提供的能 Snapshot 读历史读增量备份同个 Tablet 内多写的原性个写操作的执步骤 leader 收到写请求, 获得要写的锁为写操作分配 timestamp 通过 raft 协议备份写操作真正对数据做更改更改 timestamp 为 committed, 对外可见

16 MVCC & 数据存储 Insert: 所有新插的数据均存储在 MemRowSet 中插前使用 interval tree 和 bloom filter 检查是否已存在 Update/Delete: 对 MemRowSet 中的数据更新,Flush 后形成 Undo Records 对 DiskRowSet 中的数据更新, 存储在 DeltaMemStore 中,Flush 后形成 Redo Records Tablet Server 内存 Tablet MemRowSet DeltaMemStore 磁盘 WAL DiskRowSet (BaseData) Undo Redo Redo MemRowSet Flush DeltaMemStoreFlush

17 Flush 示例 insert value = 1 set value = 2 MemRowSet delete insert value = 4 MemRowSet Flush insert key= a, value=1 // t1 update set value=2 where key= a // t2 delete where key= a // t3 insert key= a, value=4 // t4 // MemRowSet Flush update set value=5 where key= a // t5 Undo Records Base Data delete set value = 1 insert value = 2 delete value = 4 DeltaMemStore Flush Undo Records Base Data Redo Records delete set value = 1 insert value = 2 delete value = 4 value = 5 t0 null t1 val =1 t2 val =2 t3 null t4 val =4 时间

18 Delta Compaction Minor Compaction: 减少 Redo 件的数量, 增加读效率 Major Compaction: 只读 Base Data, 节省 apply delta 的操作 Undo Base Data Redo Redo Minor Compaction Undo Base Data new Redo Major Compaction new Undo new Base Data

19 DiskRowSet Compaction 减少 DiskRowSet 间的重叠情况写操作时减少需要查询的 DiskRowSet 读操作读取更少的件, 增加读效率后台线程在空闲时做小规模 Compaction 在设计表的 schema 时, 需要考虑尽量进顺序写, 以减少后期 Compaction 的操作不要使用 id 做为主键的第列 Compaction 前 (1, a ) (3, c ) (2, b ) (5, e ) (4, d ) (7, g ) (6, f ) (8, h ) Compaction 后 (1, a ) (2, b ) (3, c ) (4, d ) (5, e ) (6, f ) (7, g ) (8, h ) Key Range

20 Integration with Spark Kudu 为 Spark 带来了什么? 实时数据的快速分析谓词下推, 加速查询基于主键索引的快速查询 Update 和 Delete 的支持 Spark 为 Kudu 带来了什么? 更简单的数据操作式

21 NoSQL API Write KuduTable table = client.opentable( my_table ); KuduSession session = client.newsession(); Insert ins = table.newinsert(); ins.getrow().addstring( metric, load-avg.1sec ); ins.getrow().adddouble( value, 0.05); session.apply(ins); session.flush(); Read KuduScanner scanner = client.newscannerbuilder(table).setprojectedcolumnnames(lists.of( value )).build(); while (scanner.hasmorerows()) { RowResultIterator batch = scanner.nextrows(); while (batch.hasnext()) { RowResult result = batch.next(); System.out.println(result.getDouble( value )); } }

22 Spark DataFrame // spark-shell --packages org.apache.kudu:kudu-spark_2.10:1.2.0 // Import kudu datasource import org.kududb.spark.kudu._ val kududataframe = sqlcontext.read.options( Map("kudu.master" -> "master1,master2,master3", "kudu.table" -> "my_table_name")).kudu // Then query using Spark data frame API kududataframe.select("id").filter("id" >= 5).show() // (prints the selection to the console) // Or register kududataframe as a table and use Spark SQL kududataframe.registertemptable("my_table") sqlcontext.sql("select id from my_table where id >= 5").show() // (prints the sql results to the console)

23 KuduRDD 的实现 getpartitions 根据下推的 predicate 获得需要访问的 tablet list 每个 tablet 均包装为一个 Partition getpreferredlocations 每个 tablet 的 replica locations execute 新建 KuduScanner 进行数据扫描

24 SparkSQL on Kudu 目标 : 和 SparkSQL on HDFS 提供致的对外接 ( 数据库 / 表 /SQL) 更好的利用 Kudu 提供的能, 如 update/delete 增加内部需要的功能, 例如动态分区管理, 权限控制等实现 : 修改 SparkSQL 语法增加 KuduExternalCatalog, 直接使用 KuduClient 获得表的元数据信息新建 Kudu 系统表, 存储内部功能需要的其他元信息

25 SparkSQL on Kudu // 创建表 CREATE TABLE kudu_test ( year INT WITH (key=true), id INT WITH (key=true), msg STRING ) PARTITIONED BY HASH (id) INTO 10 BUCKETS, RANGE (year) RANGE BETWEEN ((2015),(2016)); // 增删分区 ALTER TABLE kudu_test ADD PARTITION ((2016), (2017)); ALTER TABLE kudu_test DROP PARTITION ((2015), (2016)); // 展示 Range 分区 SHOW PARTITIONS kudu_test; // Update/Delete UPDATE kudu_test SET msg= b WHERE year=2015 AND id=1; DELETE FROM kudu_test WHERE year=2015 AND id=1;

26 Twitter 实时数据 13M 数据当前时间最新记录 Adhoc 查询

27 小米的实时数据仓库 Kudu 数据源消息队列 Spark Streaming SparkSQL HDFS

28 限制 & Roadmap 限制 : 没有自增主键不支持嵌套结构监控具不完善 Load balance Roadmap: 运维上, 增强运维具的功能 ( 错误恢复, 问题诊断 ) 性能上, 和查询引擎的深度优化 (runtime filter), 优化批量导的性能功能上, 完善安全机制, 细粒度的权限管理, 和 Sentry 的整合

29 Kudu vs 流式计算流式计算的些局限 : 预计算固定的指标,Ad-hoc 查询能较弱复杂的查询难以支持 ( 表 join) 精度要求 (TopN,count(distinct)) 存在时间窗容错状态保存, 依赖于外部存储, 难以开发

30 Kudu vs Druid 数据精度不同 : Druid 对数据进预聚合, 不保存原始数据内部存储不同 : 均是列存储, 但 Druid 内部使用 bitmap 倒排表,Kudu 使用 bitshuffle 等 encoding 式应用场景不同 : Druid 更适合做在线指标的实时统计等作,Kudu 适合实时的分析类查询查询式不同 : Druid 有自的查询引擎, 尚不支持 join;kudu 依靠 Impala/SparkSQL, 对 SQL 的支持更加完整查询性能不同 : Druid 查询的是 OLAP Cube, 速度更快 ;Kudu 需要扫描原始数据, 可支持的查询更灵活

支付宝2011年 IT资产与费用预算

支付宝2011年 IT资产与费用预算 OceanBase 支持 ACID 的可扩展关系数据库 qushan@alipay.com 2013 年 04 月关系数据库发展 1970-72:E.F.Codd 数据库关系模式 20 世纨 80 年代第一个商业数据库 Oracle V2 SQL 成为数据库行业标准可扩展性 Mainframe: 小型机 =>