PGConfChina2016-Kudu

Size: px

Start display at page:

Download "PGConfChina2016-Kudu"

流汪
4 years ago
Views:

1 Hadoop 最新结构化存储利器 Kudu 陈飚 Cloudera

关于我陈飚 Cloudera 售前技术经理资深方案架构师 http://biaobean.

2 关于我陈飚 Cloudera 售前技术经理资深方案架构师原 Intel Hadoop 发行版核心开发人员, 成功实施并运维多个上百节点 Hadoop 大数据集群曾在 Intel 编译器部门从事服务器中间件软件开发, 擅长服务器软件调试与优化, 与团队一起开发出世界上性能领先的 XSLT 语言处理器 2010 年后开始 Hadoop 产品开发及方案顾问, 先后负责 Hadoop 产品化 HBase 性能调优, 以及行业解决方案顾问 2

3 Kudu: 效分析快数据的存储系统 DATA ENGINEERING DATA DISCOVERY & ANALYTICS DATA APPS BATCH STREAM SQL SEARC H SPARK, HIVE, PIG UNIFIED DATA SERVICES RESOURCE MANAGEMENT YARN SECURITY SENTRY MODEL ONLIN E SPARK IMPALA SOLR SPARK HBASE Hadoop 生态中新的支持数据修改的列式存储系统简化可变数据上构建分析应用的架构旨在提升快速分析的性能与 Hadoop 原生集成 FILESYSTEM HDFS DATA INTEGRATION & STORAGE RELATIONAL KUDU INGEST SQOOP, FLUME, KAFKA NoSQL HBASE Apache 顶级项目,100% 开源 2016 年 9 月发布 1.0 正式版本

4 开发 Kudu 的初衷和设计目标 Why build Kudu?

5 为什么 Cloudera 要开发 Kudu? 有没有因为 Hadoop 生态系统中缺少某种存储技术而使我们无法解决客户面临的业务问题? 我们的大数据平台充分利用了过去 10 年硬件的进步吗?

6 硬件行业的进展机械硬盘 -> 固态硬盘 NAND 闪存 : 读数据时 IOPS 达到 45 万个, 写数据时 IOPS 达到 25 万个, 读数据的吞吐率约为每秒 2GB, 写数据的吞吐率达到每秒 1.5GB, 存储成本低于每 GB3 美元并且持续下降 3D XPoint memory ( 比 NAND 快 1000 倍, 比 RAM 更有成本优势 ) 内存成本更低, 单机内存量更大 : 过去几年 64->128->256GB 启示 1: CPU 将成为性能瓶颈, 现有存储系统设计并未考虑高效使用 CPU 启示 2: 列式存储也可以用于随机访问

7 联分析开发员希望知道如何优化充电性能新版本软件升级后随着时间推移是如何影响汽性能的? 实时客户希望知道是否是未成年在驾驶他们加速多快? 时速多少? 他们在哪? 汽设备如在服务前或服务中拿到最新的诊断数据包

8 联分析开发员希望知道如何优化充电性能快速有效率的 Scan 新版本软件升级后随着时 = HDFS 间推移是如何影响汽性能的? 实时客户希望知道是否是未成年在驾驶他们加速多快快速插入? 时速多少 / 查找? 他们在哪? = HBase 汽设备如在服务前或服务中拿到最新的诊断数据包

know what is the location and speed of their car in real-time.

9 How would we build the Real-time System Today? Cars are constantly producing data, being ingested in real-time or micro-batches A consumer wants to know what is the location and speed of their car in real-time. Requires: potentially millions of low latency writes / second low latency, random access Hadoop 12:01:05 AM: Where is my car? How fast? 12:01:05 AM: Your car is located at. 12:01:05 AM: Here I am! Consumer

10 How would we build the Real-time System Today? Hbase Provides: Fast, Random Read & Write Access Mini-scans Scale-out architecture capable of serving Big Data to many users HBas e Connected Cars Events Kafka / Pub-sub Consumer

11 How would we build the Analytics System Today? Cars are constantly producing data, being ingested in real-time or micro-batches GM just made a software update to temp control module. How s it working? Requires fast, efficient scan performance Hadoop 12:01:05 PM: Here I am! For the last 3 days, give me engine temp vs. radiator data. Here you go. Analyst

12 How would we build the Analytics System Today? HDFS Excels at: Full table scans Ad-hoc analytics Historic Data Events Connected Cars Kafka / Pub-sub 1. Have we accumulated enough data? Hbase/Fl ume 2. Flush into HDFS Yesterday s Partition Today s Partition Analyst

13 Hybrid big data analytics pipeline Before Kudu HBas e Random Reads Consumer Connected Cars Events Kafka / Pub-sub Snapshot & Convert to Parquet HDFS (Storage) Analytics Analyst Compact late arriving data

14 HBase+HDFS 混合架构的复杂性同时提供高性能的顺序扫描和随机查询, 避免使用 HBase+HDFS 混合架构的复杂性 : 开发 : 必须编写复杂的代码来管理两个系统之间的数据传输及同步运维 : 必须管理跨多个不同系统的一致性备份安全策略以及监控业务 : 新数据从达到 HBase 到 HDFS 中有时延, 不能马上供分析在实际运行中, 系统通常会遇到数据延时达到, 因此需要对过去的数据进行修正等如果使用不可更改的存储 ( 如 HDFS 文件 ), 将会非常不便

15 Hybrid big data analytics pipeline After Kudu Kudu supports simultaneous combination of sequential and random reads and writes Connected Cars Events Kafka / Pub-sub Kudu Random Reads Analytics Consume r Analyst

16 What is Kudu?

17 当前 Hadoop 平台上的存储组件概括 HDFS 的强项 : 效的顺序扫描能持吞吐的数据追加 HBase 的强项 : 效的按随机存取能持数据的修改可以和熊掌兼得吗? 如何实现对实时变化的数据集做效的数据分析呢 (Fast Analysis on Fast Data)?

18 Kudu 的设计目标扫描数据量时吞吐率 ( 列式存储和多副本机制 ) 目标 : 相对 Parquet 的扫描性能差距在 2x 之内访问少量数据时延时低 ( 主键索引和多数占优复制机制 ) 目标 : SSD 上读写延时不超过 1 毫秒类似的数据库语义 ( 初期持单记录的 ACID) 关系数据模型 SQL 查询 NoSQL 格的扫描 / 插 / 更新 (Java 客户端 )

19 Kudu: 可扩展的高速结构化存储可扩展性通过 275 台服务器测试 ( 集群数据量约 3PB) 设计能持扩展到上千节点的服务器和 PB 的数据量速集群吞吐率达到每秒上百万读 / 写每节点数据读取吞吐率每秒 GB 表如同传统数据库样结构化表来表数据单表记录数可超过千亿条 19

20 Kudu 表中的行记录访问类似 SQL 模式的表有限的列数 ( 不同于 HBase/Cassandra) 数据类型 : BOOL, INT8, INT16, INT32, INT64, FLOAT, DOUBLE, STRING, BINARY, TIMESTAMP 一部分列构成联合主键快速 ALTER TABLE NoSQL 风格的 Java, Python 和 C++ 语言 API Insert(), Update(), Delete(), Scan() 与 MapReduce, Spark 和 Impala 无缝对接正在对接更多处理引擎, 如 Drill, Hive

21 Kudu 不是不是个 SQL 引擎, Bring Your Own SQL ( BYO SQL ) Kudu 是存储层,SQL 处理层需要 Impala 或 Spark 等不是个件系统数据必须按表的形式结构化存储不是个运在 HDFS 上的应 Kudu 是另外个原的 Hadoop 存储引擎通常是要和 HDFS 并肩同部署不是个内存数据库能很快的处理数据可全部载入内存的任务, 也能支持非常大的数据量! 不是来替代 HDFS 或 HBase 要为适当的应场景选择适当的存储在很多应场景下 HDFS 或 HBase 仍然会更适合 Cloudera 会继续耕耘所有这些组件

22 Kudu+Impala 与 MPP DWH 的较共性通过 SQL 进快速分析查询, 并包含多数常的最新特性能插更新和删除数据不同更快的流式数据插更好的 Hadoop 集成在同集群中能进 HDFS 和 Kudu 表之间的 JOIN 与 Spark, Flume 等的集成批量插变慢没有基于事务的数据加载跨事务以及索引

23 Kudu 应场景

24 Kudu 的应场景 Kudu 最适合需要同时持顺序和随机读和写的应场景时间序列举例 : 股市情数据 ; 欺诈检测和预防 ; 险监控应类型 : Insert, updates, scans, lookups 机器数据分析举例 : Network threat detection 应类型 : Inserts, scans, lookups Online Reporting 举例 : ODS 应类型 : Inserts, updates, scans, lookups

25 业界应 Financial Services Retail Public Sector Streaming market data Real-time fraud detection & prevention Risk monitoring Real-time offers Location-based targeting Geospatial monitoring Risk and threat detection (real time)

26 当前 Hadoop 实时数据分析的现状当前的大数据架构 : 存储架构太复杂积累了足够数据吗? 流入数据 ( 消息系统 ) HBase Impala on HDFS 历史数据报表需求但是怎样处理下面的问题? 怎么有效处理转换过程中的错误? 如何定义将 HBase 数据转换成 Parquet 格式作业的周期? 将 Hbase 存储的数据重新组织为 Parqet 格式 Parquet 文件最近的数据分区最新的数据分区等待数据文件的生成在 Impala 中构建新的分区并导入数据从数据进入到报表中能体现之间的时延如何量化? 作业流程怎么保障不被其他操作打断?

27 使用 Kudu 的 Hadoop 实时数据分析流入数据 ( 消息系统 ) Storage in Kudu 历史和实时数据改进点 : 只要一套系统不需要后台定时的批处理任务报表需求轻松应对数据迟到和数据修正新数据立即用于在分析和业务运营

28 Design & Internals

29 Kudu 的核心设计有类型的存储基础构件 : 子表 (Tablet) 数据表被水平划分为子表, 类似于分区 (Partition) 通过类似于 Paxos 的投票协议 (Raft) 来管理一致性框架支持地理上的分开部署, 支持多活 (Active/Active) 系统

30 Tables( 表 ) 和 Tablets( 子表 ) 表被水平划分为子表支持按范围或哈希分区 PRIMARY KEY (host, metric, timestamp) DISTRIBUTE BY HASH(timestamp) INTO 100 BUCKETS 每个子表有多个副本 (3 个或 5), 采用 Raft 一致性协议数据可以从任一副本读取, 并且数据由 Leader 更新保证, 自动容错低 MTTR: ~5 seconds 子表服务器保存子表数据存放在本地, 而不是 HDFS Master 节点负责元数据管理服务创建 / 删除表和子表 (Tablet) 定位子表 (Tablet)

31 元数据带副本的 master* 用作子表目录 ( META 表 ) 用作 catalog (table schemas, etc) 用于负载均衡 ( 跟踪子表服务器是否在线, 增加低于要求副本数的子表的副本数 ) 所有元数据都缓存在内存, 性能更好 80 节点的压力测试, GetTableLocations RPC 性能 : 99 th percentile: 68us, th percentile: 657us <2% peak CPU usage 在客户端上配置 master 地址首次访问时向 master 询问子表地址, 随后缓存在客户端

33 LSM vs Kudu LSM Log Structured Merge (Cassandra, HBase, etc) Inserts 和 updates 先都写内存 (MemStore) 然后再刷到磁盘件 (HFile/SSTable) 读数据时合并磁盘 HFile 得到结果 Kudu 有些共同点 (Memstore, Compaction) 实现更精巧为了更好的读 ( 特别是扫描 ) 性能牺牲少许写性能

34 性能特性非常高效率地利用 CPU 使用最新特性的 C++ 实现, 包括特殊 CPU 指令 LLVM 的 JIT 编译等技术延时取决于存储硬件的性能在 SSD 或更新技术上的响应延时一般不超过 1 毫秒没有内存 GC, 使用大内存也不会有暂停使用 Bloom Filter 减少了很多磁盘访问的要求

35 Kudu 的折衷处理随机更新速度稍慢 HBase 的实现模型需任何磁盘访问就能完成随机的数据更新 Kudu 在更新时需要次主键查找, 在插时需要 Bloom 查询次, 有可能需要到磁盘上寻址单读也可能稍慢列式存储更有利于扫描 (Scan) 场景在读取最近更新过很多次的行时尤其慢 ( 例如 YCSB 的 zipfian 测试 ) 未来 : 可能引入列组 (Column Group) 来解决更注重单行读取的场景

36 容错 FOLLOWER 临时故障 : Leader 仍旧可以获得多少投票在 5 分钟内重启子表服务器, 子表服务器会重新加入集群, 对用户透明 LEADER 临时故障 : Leader 每 1.5 秒应该向 Followers 发生心跳信息如果连续 3 次没收到 Leader 的心跳信息将触发新一轮的 Leader 选举 3! 新 LEADER 在几秒钟内选出旧 Leader5 分钟内重启后以 FOLLOWER 身份重新加入 N 个副本可以容忍 (N-1)/2 个节点失败

37 容错 (2) 长期失败 : Leader 感知到某个 follower 失败超出 5 分钟排除该 follower Master 选择一个新副本 Leader 将数据拷贝到该新副本

38 子表的设计先将 Inserts 写入内存 ( 类似于 HBase 的 Memstore) 然后刷到磁盘列式存储, 与 Apache Parquet 格式类似 Updates 采用 MVCC ( 更新都带有时间戳, 不会直接修改原始数据 ) 允许 SELECT AS OF <timestamp> 查询和跨子表的一致性扫描扫描当前数据, 性能接近最优无需按行进行分支, 快速的矢量化解码和谓词计算如果对最新数据执行高频率的更新, 性能下降

39 APIs

40 原生支持的客户端 First class clients implemented in Java and C++ Experimental Cython client that wraps the C++ Supported operations DDL (create, alter, delete tables) DML (insert, update, delete) Scan (with predicate pushdown, column projection)

41 Impala + Kudu CREATE TABLE my_first_table ( id BIGINT, name STRING ) DISTRIBUTE BY HASH (id) INTO 16 BUCKETS TBLPROPERTIES( 'storage_handler' = 'com.cloudera.kudu.hive.kudustoragehandler', 'kudu.table_name' = 'my_first_table', 'kudu.master_addresses' = 'kudu-master.example.com:7051', 'kudu.key_columns' = 'id' );

42 Impala + Kudu Pre Splitting CREATE TABLE cust_behavior ( _id BIGINT, salary STRING, edu_level INT, usergender STRING, rating INT) DISTRIBUTE BY RANGE(_id) SPLIT ROWS(( ), ( ), ( ), ( ), ( ), ( )) TBLPROPERTIES( 'storage_handler' = 'com.cloudera.kudu.hive.kudustoragehandler', 'kudu.table_name' = 'my_first_table', 'kudu.master_addresses' = 'kudu-master.example.com:7051', 'kudu.key_columns' = 'id' );

43 Impala + Kudu Update & Delete UPDATE my_first_table SET name="bob" where id = 3; DELETE FROM my_first_table WHERE id < 3;

44 Impala integration CREATE TABLE DISTRIBUTE BY HASH(col1) INTO 16 BUCKETS AS SELECT FROM INSERT / UPDATE / DELETE Optimizations: predicate pushdown, scan locality, scan parallelism More optimizations on the way Not an Impala user? Community working on other integrations (Hive, Drill, Presto, etc)

45 Spark DataSource integration Available as of Kudu 0.9 // Import kudu datasource import org.kududb.spark.kudu._ val kududataframe = sqlcontext.read.options( Map("kudu.master" -> "master.address.example.com", "kudu.table" -> "my_table_name")).kudu // Then query using spark api or register a temporary table and use spark sql kududataframe.select("id").filter("id" >= 5).show() // (prints the selection to the console) // Register kududataframe as a temporary table for spark-sql kududataframe.registertemptable("kudu_table") // Select from the dataframe sqlcontext.sql("select id from kudu_table where id >= 5").show() // (prints the sql results to the console)

46 TPC-H (Analytics benchmark) 集群由 75 个 TS 和一个 master 构成每个节点 12 块硬盘, 128GRAM Kudu Impala 2.2+CDH 5.4 TPC-H Scale Factor 100 (100GB) 结果对内存数据,Kudu 性能比 Parquet 高 31% ( 几何平均 ) 对硬盘数据,Parquet 性能应该比 Kudu 更好 (larger IO requests

47 用户案例

48 大数据分析处理流程使用 Kudu 之前在使用 Kudu 前, 小米的大数据分析 Pipepline 主要有几种 : 1. 数据源 -> Scribe 将日志输出到 HDFS -> MR/Hive/Spark -> HDFS Parquet -> Impala -> 将结果对外服务, 这个数据流一般用于分析各种日志 2. 数据源 -> 实时更新 HBase/MySQL -> 每天批量导出 Parquet-> Impala -> 将结果对外服务, 这个数据流一般用来分析状态数据, 也就是一般需要随机更新的数据比如用户 Profile 之类的这两条数据流主要有几个问题 : 1. 数据从生成到落地成能被高效查询的列式存储, 整个延时比较大, 一般是小时级别到一天 2. 很多数据的日志到达时间和逻辑时间不一致, 一般存在一些随机延时比如很多 mobile app 统计应用, 这些 tracing event 发生后很可能过一段时间才被后段 tracing server 收集到对于一些实时分析需求, 有一些可以通过流处理来解决, 不过没有 SQL 方便, 另外流式处理只能做固定的数据分析, 对 ad-hoc 查询无能为力 Kudu 的特点正好可以来配合 Impala 搭建实时 ad-hoc 分析应用

49 大数据分析处理流程使用 Kudu 简化改进后的数据流大概是这样 : 1. 数据源 -> Kafka -> ETL(Storm) -> Kudu -> Impala 2. 数据源 -> Kudu -> Impala 数据流 1 主要是为需要进一步做 ETL 的应用使用的, 另外 Kafka 可以作为 Buffer, 当写吞吐有毛刺时,Kafka 可以做一个缓冲如果应用有严格的实时需求, 就是只要数据源写入就必须能够查到, 就需要使用数据流 2

来自小米的基准测试从程序跟踪分析应用中抽取的 6 个真实的查询场景 Q1: SELECT COUNT(*) Q2: SELECT hour, COUNT(*) WHERE module = foo GROUP BY HOUR Q3: SELECT hour, COUNT(DISTINCT uid) WHERE module = foo AND app= bar GROUP BY HOUR

50 来自小米的基准测试从程序跟踪分析应用中抽取的 6 个真实的查询场景 Q1: SELECT COUNT(*) Q2: SELECT hour, COUNT(*) WHERE module = foo GROUP BY HOUR Q3: SELECT hour, COUNT(DISTINCT uid) WHERE module = foo AND app= bar GROUP BY HOUR Q4: analytics on RPC success rate over all data for one app Q5: same as Q4, but filter by time range Q6: SELECT * WHERE app = AND uid = ORDER BY ts LIMIT 30 OFFSET 30 测试集群 :71 台服务器硬件 :CPU: E GHz * 24 core Memory: 64GB Network: 1Gb Disk: 12 HDD 软件 :Hadoop2.6/Impala 2.1/Kudu 数据量 : 一天的服务器端跟踪日志数据约 26 亿条数据每条数据约 270 字节表有 17 个字段, 其中 5 个为主键字段 50

测试结果 Bulk load using impala (INSERT INTO): Total Time(s) Throughput(Total) Throughput(per node) Kudu 961.1 2.8M record/s 39.5k record/s Parquet 114.

51 测试结果 Bulk load using impala (INSERT INTO): Total Time(s) Throughput(Total) Throughput(per node) Kudu M record/s 39.5k record/s Parquet M record/s 331k records/s Query latency (seconds): HDFS parquet file replication = 3 Kudu table replication = 3 Each query run 5 times then averaged

52 京东案例 Jd.com 中国第在线电商使 Kafka 实时收集数据点击流志应 / 浏览器 Trace 志每条记录约 70 字段 6/18 sale day 150 亿笔交易峰期每秒千万条数据插集群 200 台服务器查询使 JDBC -> Impala -> Kudu Browser tracing Kafka Kudu Impala JDBC access Web-app Marketing Dept. Web logs Developers 52

53 怎样开始?

54 Apache Kudu 社区

55 作为用户 Quickstart VM 轻松上手! Impala and Kudu in an easy-to-install VM CSD and Parcels For installation on a Cloudera Manager-managed cluster 白皮书 :getkudu.io/kudu.pdf

56 作为开发人员所有的代码首先都会提交到这里 Public gerrit: 所有的代码审查会提交到这里 Public JIRA: 自 2013 以来的缺陷修复. Come see our dirty laundry! Apache 2.0 开源代码许可欢迎大家贡献代码和功能!

58 陈飚

ABOUT ME AGENDA 唐建法 / TJ MongoDB 高级方案架构师 MongoDB 中文社区联合发起人 Spark 介绍 Spark 和 MongoDB 案例演示

完整的大数据解決方案 ABOUT ME AGENDA 唐建法 / TJ MongoDB 高级方案架构师 MongoDB 中文社区联合发起人 Spark 介绍 Spark 和 MongoDB 案例演示 Dataframe Pig YARN Spark Stand Alone HDFS Spark Stand Alone Mesos Mesos Spark Streaming Hive Hadoop