Jacky Li / 李昆 Huawei BigData Platform Engineer Apache CarbonData PMC & Committer Previous experience: L3/L4/L7 Network, Protocol, Security Telecom Val

Size: px

Start display at page:

Download "Jacky Li / 李昆 Huawei BigData Platform Engineer Apache CarbonData PMC & Committer Previous experience: L3/L4/L7 Network, Protocol, Security Telecom Val"

秤睦二申
5 years ago
Views:

1 CarbonData: 面向交互式分析的索引文件格式

2 Jacky Li / 李昆 Huawei BigData Platform Engineer Apache CarbonData PMC & Committer Previous experience: L3/L4/L7 Network, Protocol, Security Telecom Value Added Service Platform Network Data Analytics 2

3 Big Data 企业数据量大, 维度多, 结构复杂, 且在快速增长 Network 54B records per day 750TB per month Complex correlated data Consumer 00 thousands of sensors >2 million events per second Time series, geospatial data Enterprise 00 GB to TB per day Data across different domains 3

4 Typical Scenario 企业中包含多种数据应用, 从商业智能到批处理到机器学习 Report & Dashboard OLAP & Ad-hoc Batch processing Machine learning Realtime Analytics data Big Table Ex. CDR, transaction, Web log, Small table Small table Unstructured data 4

5 Analytic Examples 过去天使用 Whatapp 应用的终端按流量排名情况? 过去天上海市每个小区的网络拥塞统计? Tracing and Record Query for Operation Engineer 5

6 Challenge - Data Data Size Single Table >0 B Fast growing Multi-dimensional Every record > 00 dimension Add new dimension occasionally 百亿级数据量多维度 Rich of Detail Billion level high cardinality B terminal * 200K cell * 440 minutes = ( 万亿 ) 细粒度 6

7 Challenge - Application Enterprise Integration 企业应用集成 SQL 2003 Standard Syntax BI integration, JDBC/ODBC Multi-dimensional OLAP Query Flexible Query Any combination of dimensions OLAP Vs Detail Record Full scan Vs Small scan 灵活查询无固定模式 Precise search & Fuzzy search Full Scan Query Small Scan Query 7

8 How to choose storage? 如何构建数据平台? 8

9 Option: NoSQL Database Key-Value store: low latency, <5ms 只能通过 Key 访问, 一键一值适合实时应用对接, 不适合分析型应用 9

10 Option2:Parallel database Parallel scan + Fast compute 细粒度控制并行计算, 适合中小规模数据分析 ( 数据集市 ) Questionable scalability and fault-tolerance 0 Cluster size < 00 data node Not suitable for big batch job 扩展能力有上限查询内容错能力弱不适合海量数据分析 ( 企业级数仓 )

11 Option3: Search engine All column indexed Fast searching Simple aggregation 适合多条件过滤, 文本分析 Designed for search but not OLAP Not for TopN, join, multi-level aggregation 3~4X data expansion in size No SQL support 无法完成复杂计算数据膨胀专用语法, 难以迁移

12 Option4: SQL on Hadoop Modern distributed architecture, scale well in computation. Pipeline based: Impala, Drill, Flink, BSP based: Hive, SparkSQL 并行扫描 + 并行计算适合海量数据计算 BUT, still using file format designed for batch job 仍然使用为批处理设计的存储, 场景受限 Focus on scan only No index support, not suitable for point or small scan queries 2

13 Capability Matrix 只针对某个场景设计, 解决一部分问题 Option Store Good Bad Multi-dimensional OLAP Query KV Store Parallel database Search engine HBase, Cassandr a, Greenplum, Vertica, Solr, ElasticSearc h, Full Scan Query Small Scan Query SQL on Hadoop - Pipeline SQL on Hadoop - BSP Impala, HAWQ, Drill, Hive, SparkSQL 3

14 Architect s choice Choice : Compromising 做出妥协, 只满足部分应用 App App2 App3 Choice 2: Replicating of data 复制多份数据, 满足所有应用 App App2 App3 Loading Data Replication Data 4

15 Motivation Multi-dimensional OLAP Query CarbonData: Unified Storage Full Scan Query Small Scan Query 一份数据满足多种分析场景详单过滤, 海量数仓, 数据集市, 5

16 Spark + CarbonData: 打造大数据交互式分析引擎 6

17 Apache CarbonData 社区介绍 CarbonData 206 年 6 月全票通过正式进入 Apache 孵化器目标 : Compute 更易用, 一份存储覆盖更多场景更高的分析性能, 面向用户提供交互式分析已发布了 3 个 Apache 稳定版本欢迎订阅邮件列表和贡献 : Storage Code: JIRA: Maillist: dev@carbondata.incubator.apache.org 贡献者来自 : Huawei, Talend, Intel, ebay, Inmobi, 美团, 阿里, 乐视,Hulu 7

18 Carbon-Spark Integration Built-in Spark integration Spark.5,.6,2.0 Interface SQL DataFrame API Data Management Bulk load/incremental load Delete load Compaction Data Management Reader/Writer Query Optimization Carbon File Carbon File Carbon File 8

19 Integration with Spark Query CarbonData Table DataFrame API carboncontext.read.format( carbondata ).option( tablename, table ).load() With late decode optimization and carbon-specific SQL command support sqlcontext.read.format( carbondata ).load( path_to_carbon_file ) SQL CREATE TABLE IF NOT EXISTS T (name String, PhoneNumber String) STORED BY carbondata LOAD DATA LOCAL INPATH path/to/data INTO TABLE T 9

20 Data Ingestion Bulk Data Ingestion Load from CSV file Load from other table LOAD DATA [LOCAL] INPATH 'folder path' [OVERWRITE] INTO TABLE tablename OPTIONS(property_name=property_value,...) INSERT INTO TABLE tablennme select_statement FROM table; Save Spark Dataframe as Carbon data file df.write.format( carbondata").options("tablename, tbl")).mode(savemode.overwrite).save() 20

21 Segment Introduction Every data load becomes one segment in CarbonData table, data is sorted within one segment. JDBCServer (Load) Segment Manager (ZK based) JDBCServer (Query) Carbon Table Segment Carbon File Segment Carbon File Segment Carbon File 2

22 CarbonData Table Organization Spark Index In Memory B Tree HDFS /tablename/fact/segmentid /tablename/meta Carbon File Carbon File Carbon File Carbon File Index File Dictionary File Schema File Data Footer Data Footer Data Footer Data Footer All Footer Dictionary Map Latest Schema (Index is stored in the footer of each data file) (append only) (rewrite) 22

23 Data Compaction Data compaction is used to merge small files Re-clustering across loads for better performance Two types of compactions supported - Minor compaction Compact adjacent segment based on number of segment Automatically trigger - Major compaction Compact segments based on size Manually trigger ALTER TABLE [db_name.]table_name COMPACT MINOR/MAJOR 23

24 CarbonData File Structure Built-in Columnar & Index Store index and data in the same file, co-located in HDFS Balance between batch and point query Index support: Multi-dimensional Index (B+ Tree) Min/Max index Inverted index Encoding: Dictionary, RLE, Delta Snappy for compression Data Type: Primitive type and nested type Schema Evolution: Add, Remove, Rename columns CarbonData File Blocklet Blocklet Blocklet Footer 24

25 Index Introduction Multi-level indexes: Table level index: global B+ tree index, used to filter blocks Catalyst Spark Driver Table Level Index Executor Executor File level index: local B+ tree index, used to filter blocklet Column level index: inverted index within column chunk File Level Index & Scanner Carbon File Data Footer Carbon File Data Footer File Level Index & Scanner Carbon File Data Footer Column Level Index 25

26 Encoding Example 26 Data are sorted along MDK (multidimensional keys) Data stored as index in columnar format Blocklet Logical View C C2 C3 C4 C5 C6 C Column split Sorted MDK Index [,,,,] : [42,432] [,,,,3] : [443,44622] [,,,3,2] : [54,54702] [,,2,,4] : [545,5887] [,,2,,5] : [675,568] [,,3,,7] : [570,508] [,,3,2,8] : [56,55245] Years Quarters Months Territory Country Quantity Sales 2003 QTR Jan EMEA Germany 42, QTR Jan APAC China 54 54, QTR Jan EMEA Spain , QTR Feb EMEA Denmark , QTR Feb EMEA Italy , QTR Mar APAC India 52 9, QTR Mar EMEA UK 570 5, QTR Mar Japan Japan 56 55, QTR2 Apr APAC Australia , QTR2 Apr EMEA Germany 44,532 Sort (MDK Index) Dictionary Encoding [,,,,] : [42,432] [,,,3,2] : [54,54702] [,,,,3] : [443,44622] [,,2,,4] : [545,5887] [,,2,,5] : [675,568] [,,3,3,6] : [52,9749] [,,3,,7] : [570,508] [,,3,2,8] :

27 File Level Blocklet Index Blocklet Blocklet Blocklet Blocklet Blocklet File Footer Blocklet Index Blocklet Start Key End Key C(Min, Max). C7(Min, Max) Blocklet4 Start Key4 End Key4 C(Min, Max). C7(Min, Max) Build in-memory file level MDK index tree for filtering Major optimization for efficient scan Start Key End Key C(Min,Max) C7(Min,Max) Start Key End Key2 Start Key2 End Key2 C(Min,Max) C7(Min,Max) Start Key End Key4 Start Key3 End Key3 C(Min,Max) C7(Min,Max) Start Key3 End Key4 Start Key4 End Key4 C(Min,Max) C7(Min,Max) 27

28 Block Pruning Spark Driver side index (table level) Query optimization Predicate push-down: leveraging multi-level indexes Column Pruning HDFS Blocklet C C2 C3 C4 C5 C6 C7 C9 Block Block Block Block Blocklet Blocklet Blocklet Blocklet Blocklet Blocklet Blocklet Blocklet Inverted Index Footer Footer Footer Footer 28

29 Column Chunk Inverted Index Blocklet ( sort column within column chunk) Optionally store column data as inverted index within column chunk suitable to low cardinality column better compression & fast predicate filtering C Blocklet Physical View C2 C3 C4 C5 C6 C7 d r d r d r d r d r d r Column chunk Level inverted Index Dim Block (-0) Columnar Store Dim2 Block (-8) 2(9-0) [ ] :[ ] :[ ] :[ ] :[ ] : [42]:[432] [ 2] :[ 2] :[ 2] :[ 2] :[ 9] : [443]:[44622] [ 3] :[ 3] :[ 3] :[ 4] :[2 3] : [54]:[54702] [ 4] :[ 4] :[2 4] :[ 5] :[3 2] : [545]:[5887] [ 5] :[ 5] :[2 5] :[ 6] :[4 4] : [675]:[568] [ 6] :[ 6] :[3 6] :[ 9] :[5 5] : [570]:[508] [ 7] :[ 7] :[3 7] :[2 7] :[6 8] : [56]:[55245] [ 8] :[ 8] :[3 8] :[3 3] :[7 6] : [52]:[9749] [ 9] :[2 9] :[4 9] :[3 8] :[8 7] : [44]:[532] [ 0]:[2 0]:[4 0]:[3 0] :[9 0] : [525]:[50398] Dim3 Block (-3) 2(4-5) 3(6-8) 4(9-0) Run Length Encoding & Compression Blocklet Rows Dim4 Block (-2,4-6,9) 2(7) 3(3,8,0) Dim5 Block (,9) 2(3) 3(2) 4(4) 5(5) 6(8) 7(6) 8(7) 9(0) Measure Block Measure2 Block [42]:[432] [443]:[44622] [54]:[54702] [545]:[5887] [675]:[568] [570]:[508] [56]:[55245] [52]:[9749] [44]:[532] [525]:[50398]

Column Group Allow multiple columns form a column group stored as a single column chunk in rowbased format suitable to set of columns frequently fetched together saving stitching cost

30 Column Group Allow multiple columns form a column group stored as a single column chunk in rowbased format suitable to set of columns frequently fetched together saving stitching cost for reconstructing row Blocklet C C2 C3 C4 C6 C5 Col Chunk Col Chunk Col Chunk Col Group Chunk Col Chunk

31 Nested Data Type Representation Arrays Represented as a composite of two columns One column for the element value One column for start index & length of Array Struts Represented as a composite of finite number of columns Each struct element is a separate column Name Array [start,len] Ph_Number Name Array<Ph_Number> John [92,9] Sam [2,345,333] Bob [98,787] John 0,2 92 Sam 2,3 9 Bob 5, Name John Sam Bob Info Strut<age,gender> [3,M] [45,F] [6,M] Name Info.age Info.gender John 3 M Sam 45 F Bob 6 M

32 Encoding & Compression Efficient encoding scheme supported: DELTA, RLE, BIT_PACKED Dictionary: table level global dictionary -> Enable Lazy Decode optimization Compression: Column data compression: Snappy Adaptive Data Type Compression Big Win: Speedup Aggregation Reduce run-time memory footprint Enable fast distinct count 32

33 Lazy Decode Original plan Aggregation Filter Optimized plan DictionaryDecode Aggregation Filter Translate dictionary key to value Groupby on dictionary key Scan Scan 33

34 CarbonData 性能对比 (Spark-Parquet) 测试环境 p 集群 :3(Worker)+(Master),40 核, 384GB, 0G 网络带宽 p 软件 :Hadoop 2.7.2,Spark.5.2 p 数据 :0 亿记录,300 列, 原始数据.9TB.45X ~ 3X 查询特点 p Point Query: 基于主 key 过滤 p Small Scan: 包含多个列过滤 p Full Scan: 复杂聚合 Join, 无过滤条件 p OLAP Query: 同时带过滤, 聚合 34

35 CarbonData 性能对比 (Impala) 测试环境 p 集群 :3(Worker)+(Master),40 核, 384GB, 0G 网络带宽 p 软件 :Hadoop 2.7.2,Spark.5.2,Impala 2.6 p 数据 :0 亿记录,300 列, 原始数据 830G 对于小结果集的场景, impala 依赖全表扫描, 而 Carbon 使用索引查询特点 p 多维度过滤 p 多个 Join: 与多个维表 35

36 Success Case 应用案例 36

37 金融交互式分析 : 提供高性能交互式分析体验数据源批量作业数据应用明细数据查询聚合分析离线精准营销客户挑战 Impala 查询性能 (>700s, 且极易查询失败 ) 互联网数据数据平台 Manager 交互查询稳定性问题 (impala 存在挂死情况 ) impala 资源不能统一管理, 无法共享批量加工交互式分析系统管理应对方案日志数据应用系统数据数据交换平台数据导入 Hive Spark+Carbo n Yarn HDFS 安全管理批量加工 : Hive 交互式分析 : SparkSQL+CarbonData 客户价值资源采用 Yarn 统一管理, 用户可配, 可调百亿 - 千亿数据, 秒级响应 ( 半年查询 <0 秒, 年查询 <20 秒 ) 老系统 : 开源 Hive+Impala+Parquet,35 台服务器 37

38 电信详单分析 : 开源系统无法满足业务性能和稳定性要求数据源批量作业数据应用明细数据查询日志分析离线精准营销客户挑战按手机号码做分区, 分区多, 导致小文件问题非手机号码查询时, 查询性能较差, 并发下响应慢数据平台 Manager Impala 资源不能统一管理, 无法共享周期汇聚详单查询系统管理应对方案网络数据每 5 分钟入库, 200 万条每秒数据导入 Spark Spark+Carbo n Yarn HDFS 安全管理周期汇聚 : Spark 详单查询 : SparkSQL+CarbonData ( 每 5 分钟入库一批数据, 每 X 批自动 compaction) 客户价值资源采用 Yarn 统一管理, 用户可配, 可调百亿 - 千亿数据, 任意维度过滤查询, 秒级响应老系统 : 开源 Hive+Impala+Parquet,98 台服务器 38

39 What s Next 下一步计划 39

40 What s coming next: functionality Batch Update: Support daily update scenario for OLAP, Not OLTP! Streaming Ingest: Introduce row-based format for fast ingestion Compaction from row to columnar format Broader Integration across Hadoop-ecosystem: Flink, Kafka, Kylin 40

41 What s coming next: performance Better query performance by Spark 2.X integration Whole stage codegen Vectorized reader Better join performance by bucket table Better loading performance Offheap sort Single-pass loading (on-the-fly dictionary generation) 4

42 42 数据更新 ( 预览 )

43 Data Update/Delete (beta) 场景 : 缓慢变化的维表, 如刷新用户信息批量更新的事实表, 如更新某些度量值,ID 字符串 ACID 属性 : 支持更新语句原子性 ( 要么成功要么失败 ), 一致性 ( 更新成功后马上可查 ) 更新过程中不影响查询不支持并行更新, 不支持 OLTP 类应用 43

44 Data Update/Delete (beta) Delete Delta 文件 : 记录要删除的行号, bitmap 文件 Update Delta 文件 : 记录要插入的记录, Carbon 文件更新流程 :. 找到要更新的 Segment 和目标文件 2. 找到要更新的行 3. 写入 Delete Delta 和 Update Delta 文件读取流程 :. 读取 Base 文件 2. 排除 Delete 记录 (RowID) 3. 加入 Update Delta( 新记录 ) 44

45 计划在月份 CarbonData.0 中发布 JIRA: CARBONDATA-440 issues.apache.org/jira/browse/carbondata-440 github.com/apache/incubator-carbondata 45

46 Copyright 206 Huawei Technologies Co., Ltd. All Rights Reserved. The information in this document may contain predictive statements including, without limitation, statements regarding the future financial and operating results, future product portfolio, new technology, etc. There are a number of factors that could cause actual results and developments to differ materially from those expressed or implied in the predictive statements. Therefore, such information is provided for reference purpose only and constitutes neither an offer nor an acceptance. Huawei may change the information at any time without notice.

PowerPoint 演示文稿

PowerPoint 演示文稿华为最高级别技术认证演讲人 : 徐一鸣 / yiming.xu@huawei.com 职位 : HCIE 讣证系统架构师 HCIE 深度解析目录 1 2 3 4 为什么参加 HCIE 讣证 HCIE 讣证考试大揭秘怎么获得 HCIE 讣证 HCIE 相关信息获取途径为什么选择 HCIE? 知