CarbonData大数据交互式分析实践

1、CarbonData: 大数据交互式分析实践Agenda2为什么需要CarbonDataCarbonData介绍性能测试应用案例未来计划企业中包含多种数据应用，从商业智能、批 处理到机器学习Big TableEx. CDR, transaction, Web log,Small tableSmall tableUnstructured datadataReport & DashboardOLAP & Ad-hocBatch processingMachine learningRealtime Analytics3来自数据的挑战4Data SizeSingle Table 10 BFast gr

2、owingMulti-dimensionalEvery record 100 dimensionAdd new dimension occasionallyRich of DetailBillion level high cardinality1B terminal * 200K cell * 1440 minutes = 28800 (万亿）百亿级数据量多维度细粒度来自应用的挑战Enterprise IntegrationSQL 2003 Standard SyntaxBI integration, JDBC/ODBCFlexible QueryAny combination of dime

3、nsionsOLAP Vs Detail RecordFull scan Vs Small scanPrecise search & Fuzzy searchSmall Scan QueryFull Scan QueryMulti-dimensional OLAP Query企业应用集成5灵活查询 无固定模式How to choose storage?6如何构建数据平台？选择1: NoSQL DatabaseKey-Value store: low latency, 5ms只能通过Key访问，一键一值适合实时应用对接，不适合分析型应用7选择2：Parallel databaseParallel

4、 scan + Fast computeQuestionable scalability and fault-toleranceCluster size 2.2Spark 1.5, 1.6, 2.1接口SQLDataFrame API支持操作:查询：支持SparkSQL优化器数据管理：批量入库、更新、删除、合并（Compaction）、增删列Carbon FileCarbon FileCarbon FileCarbonData大数据生态18SQL19Dataframedf.write.format(“carbondata).options(tableName“, “t1).mode(SaveM

5、ode.Overwrite).save()CREATE TABLE tablename (name String, PhoneNumber String) STORED BY “carbondata”TBLPROPERTIES (.)LOAD DATA LOCAL INPATH folder path OVERWRITE INTO TABLE tablename OPTIONS(.)INSERT INTO TABLE tablennmeselect_statement1 FROM table1;使用方式：入库SQL20DataframeSELECT project_list FROM t1 W

6、HERE cond_listGROUP BY columns ORDER BY columnsdf = sparkSession.read.format(“carbondata”).option(“tableName”, “t1”).load(“path_to_carbon_file”)df.select().show使用方式：查询UPDATE table1 ASET (A.PRODUCT, A.REVENUE) =(SELECT PRODUCT, REVENUEFROM table2 BWHERE B.CITY = A.CITY AND B.BROKER = A.BROKER)WHERE A

7、.DATE BETWEEN 2017-01-01 AND 2017-01-31table1table2UPDATE table1 ASET A.REVENUE = A.REVENUE - 10WHERE A.PRODUCT = phoneDELETE FROM table1 AWHERE A.CUSTOMERID = 123123, abc456, jkdphone, 70 60 car,100 phone, 30 20Modify one column in table121Modify two columns in table1 with values from table2Delete

8、records in table1使用方式：更新和删除CarbonData介绍CarbonData文件CarbonData表格查询数据管理22CarbonData介绍CarbonData文件CarbonData表格查询数据管理23Blocklet: 文件内的数据块Data are sorted along MDK (multi-dimensional keys)Clustered data enabling efficient filtering and scanColumn chunk: Blocklet内的列数据一次IO单元内部分为多个Page，解压单元元数据和索引信息Header：Ver

9、sion，SchemaFooter： Blocklet Index & statistics内置索引Multi-dimensional IndexStatistics: Column Min/Max, CardinalityCarbon Data FileBlocklet 1Blocklet NFile FooterBlocklet Index & StatsCarbonData文件格式File HeaderVersionSchemaColumn 1 ChunkPage1Page2Page3Column 2 ChunkColumn n Chunk24YearsQuartersMonthsTer

10、ritoryCountryQuantitySales2003QTR1JanEMEAGermany14211,4322003QTR1JanAPACChina54154,7022003QTR1JanEMEASpain44344,6222003QTR1FebEMEADenmark54558,8712003QTR1FebEMEAItaly67556,1812003QTR1MarAPACIndia529,7492003QTR1MarEMEAUK57051,0182003QTR1MarJapanJapan56155,2452003QTR2AprAPACAustralia52550,3982003QTR2A

11、prEMEAGermany14411,5321,1,1,1,1 : 142,114321,1,1,3,2 : 541,547021,1,1,1,3 : 443,446221,1,2,1,4 : 545,588711,1,2,1,5 : 675,561811,1,3,3,6 :52,97491,1,3,1,7 : 570,510181,1,3,2,8 : 561,552451,2,4,3,9 : 525,503981,2,4,1,1 : 144,11532(MDSortK Index)1,1,1,1,1 : 142,114321,1,1,1,3 : 443,446221,1,1,3,2 : 54

12、1,547021,1,2,1,4 : 545,588711,1,2,1,5 : 675,561811,1,3,1,7 : 570,510181,1,3,2,8 : 561,552451,1,3,3,6 :52,97491,2,4,1,1 : 144,115321,2,4,3,9 : 525,50398数据和索引合一存储，数据即索引多维索引 (Multi-Dimensional Key)全局字典编码Blocklet (Columnar view)Sorted MDK IndexC1C2C3C4C5C6C71111114211432111134434462211132541547021121454

13、5588711121567556181113175705101811328561552451133652974912411144115321243952550398索引建立过程介绍25索引建立过程介绍（二）Blocklet 1111111120001112125000112111120001122125000113111120001132125000Blocklet 2121323110001223231100012332311000131434200013153410001324342000Blocklet 313253410001334342000133534100014141120000

14、1424112000014341120000Blocklet 4211111120002112125000212111120002122125000213111120002132125000Blocklet IndexBlocklet1Start Key1End Key1Start Key1End Key4Start Key1End Key2Start Key3End Key4Start Key2End Key2File FooterBlockletBuild in-memory file level MDK index tree for filteringMajor optimization

15、 for efficient scanC1(Min, Max).C7(Min, Max)Blocklet4Start Key1End Key1C1(Min,Max)C7(Min,Max)C1(Min,Max)C7(Min,Max)Start Key3End Key3C1(Min,Max)C7(Min,Max)Start Key4End Key4C1(Min,Max)C7(Min,Max)Start Key4End Key4C1(Min, Max).C7(Min, Max)26利用两级索引架构减少Spark Task数和磁盘IOSpark DriverExecutorDataFooterData

16、FooterCarbon FileCarbon FileCarbon FileDataFooterBlocklet IndexBlock IndexExecutorBlocklet IndexCatalyst27第一级：文件级索引用于过滤文件（HDFS Block）， 避免扫描不必要的文件，减少多 达95%的Spark Task第二级：Blocklet索引用于过滤文件内部的Blocklet， 避免扫描不必要的Blocklet，减 少磁盘IO支持嵌套数据类型：Array, StructRepresented as a composite of two columnsOne column for

17、the element valueOne column for start index & length of ArrayArrayRepresented as a composite of finite number of columnsEach struct element is a separate columnStructNameArrayJohn192,191Sam121,345,333Bob198,787NameArray start,lenPh_NumberJohn0,2192Sam2,3191Bob5,2121345333198787NameInfo StrutJohn31,M

18、Sam45,FBob16,MNameInfo.ageInfo.genderJohn31MSam45FBob16M28CarbonData介绍CarbonData文件CarbonData表格查询数据管理29CarbonData表格组成CarbonData FileMetadata:Global Dictionary, SchemaIndex & StatisticsImmutable Data FileData sourceLoad in batchSegmentCarbonData TableDelta FileDelta File for Update/DeleteIndex File30C

19、arbonData表格组成：多次入库Carbon FileDictionary (append) Schema (Overwrite)Index (new tree)Data File (new folder)Data sourceLoad in batchSegment 2Segment 1CarbonData TableDelta FileCarbon FileDelta File31CarbonData表格物理存储Carbon FileDataFooterCarbon FileDataFooterCarbon FileDataFooterCarbon FileDataFooterDict

20、ionary FileDictionary MapSpark DriverTable Level IndexIndex FileAll FooterSchema FileLatest Schema/table_name/fact/segment_id/table_name/metaHDFS32SparkSegment管理SegmentJDBCServer (Load)JDBCServer (Query)Carbon TableLeveraging HDFS File Lock to manage the Segment State, Data UpdateSegment Manager (HD

21、FS File Lock):Incremental LoadBatch UpdateCarbon FileSegmentCarbon FileSegmentCarbon File33CarbonData介绍CarbonData文件CarbonData表格查询数据管理34文件级对接：MapReduceCarbo n FileCarbon FileCarbon FileHDFSIndexInputFormat/OutputFormatSegment 3Segment 2Segment 135SQL引擎对接：表格Carbo n FileCarbon FileCarbon FileHDFSReadSQ

22、LIndexSegment 3Segment 2Segment 136SparkSQL深度对接Carbo n FileCarbon FileCarbon File与SparkSQL优化器 深度集成CarbonData TableRead/Write/Update/Delete/Alter TableSQLSegment 3Segment 2Segment 137Metadata: DictionaryIndexSparkSQL Data Source对接SQL or DataFrameResolve RelationCarbon-specific optimization rule:Lazy

23、Decode leveraging global dictionaryRule-basedNew SQL syntaxDML related statementExecutionRDDCarbonScanRDD:Leveraging multi level index for efficient filtering and scanDML related RDDsParserSparkSQL Catalyst FrameworkParser/AnalyzerOptimizerCost-basedPhysical PlanningStrategyCarbon Data Source38Spark

24、 Core利用两级索引架构减少Spark Task数和磁盘IOSpark DriverExecutorDataFooterDataFooterCarbon FileCarbon FileCarbon FileDataFooterBlocklet IndexBlock IndexExecutorBlocklet IndexCatalyst39第一级：文件级索引用于过滤文件（HDFS Block）， 避免扫描不必要的文件，减少多 达95%的Spark Task第二级：Blocklet索引用于过滤文件内部的Blocklet， 避免扫描不必要的Blocklet，减 少磁盘IOSpark Executo

25、rSpark DriverBlockletHDFSFileFooterBlockletBlockletC1C2C3C4Cn1. File pruningFileBlockletBlockletFooterFileFooterBlockletBlockletFileBlockletBlockletFooterBlockletBlockletBlockletBlockletTaskTask2. Blocklet pruningRead and decompress filter columnBinary search using inverted index, skip to next block

26、let if no matchingDecompress projection columnReturn decoded data to spark40SELECT c3, c4 FROM t1 WHERE c2=beijing一个完整的过滤查询过程大颗粒IO单元（Carbon V3格式）：Blocklet内部一个Column Chunk 是一个IO单元， Blocklet按大小切分，默认64MB，大概有100万行记 录。大颗粒顺序读提升扫描性能。跳跃解码（Carbon V3格式） ：增加数据页Page概念，按Page进行过滤和解码，减 少不必要的解码解压缩，提升CPU利用率向量化解码 + S

27、park向量化处理解码和解压缩采用向量化处理，与Spark2.1向量化、 Codegen结合，在扫描+聚合场景下提升性能4X堆外内存：处理大结果集时，解码解压缩过程中堆外完成， 减少GCHeaderVersionSchemaTimestamp向量化CarbonReaderColumnBatch结构Spark向量化处理 + Whole Stage CodeGen文件内扫描优化，与Spark向量化处理对接41Stage 2利用全局字典做延迟解码，提升GroupBy性能Final AggregationFinal AggregationStage 1Partial AggregationScan (

28、decode c2 only)Stage 2Dictionary DecodeBefore applying Lazy DecodeAfter applying Lazy DecodeSELECT c3, sum(c2) FROM t1 GROUP BY c3使用字典值（Int） 做汇聚、shuffle将字典值解码 还原为原始值（字符串）42CarbonData介绍CarbonData文件CarbonData表格查询数据管理43CREATE TABLE table(c1 STRING, c2 INT, c3 STRING, c4 INT, c5 DOUBLE)STORED BY “carbon

29、data”TBLPROPERTIES (SORT_COLUMNS=c2, c3)LOAD DATA INPATH source path INTO TABLE table OPTIONS (SORT_SCOPE=NO_SORT/BATCH_SORT/LOCAL_SORT)44增量入库：排序Key和排序范围通过SORT_COLUMNS指定需要排序的Key。如果不指定，系统会使用默认规则（查询性能最大化）：在所有维度列上建立MDK索引排序范围是LOCAL_SORT每次入库时可指定三种排序范围：NO_SORT：不排序，在Compaction时建立索引。BATCH_SORT：使用配置的内存大小进行“尽

30、量排序”LOCAL_SORT：在节点内部排序（Merge Sort）入库流水线介绍InputConvertIndex- BuildWrite读取和解析输入（CSV或RDD）字典编码转换， 嵌套类型拉平按照SORT_SCOPE进行排序列式编码、压缩、 写文件MapperSpark TaskCarbon Data Load Pipeline45全局字典的生成方式Pre-defined Dictionary入库前手工生成字典LOAD DATA path INTO TABLE t OPTIONS(COLUMNDICT=nplist)nplist := column_name:path,场景：通过静态维

31、表预生产字典Two Pass Job1：生成字典 Job2：执行入库流程场景：首次入库Single Pass由Dictionary Server在 线生成字典Dictionary Server场景：非首次入库；若维度Cardinality不高， 也可用于首次入库46数据更新和删除47场景：批量更新的事实表，如更新某些度量值，ID值缓慢变化的维表，如刷新用户信息ACID属性：支持更新语句原子性（要么成功要么失败），一致性（更新串行化）更新过程中不影响查询适用于OLAP型+批量更新场景，不支持毫秒级实时更新，不支持OLTP类应用删除数据DeleteBase FileSegmentCarbonDat

32、a TableDelete Delta48Bitmap file that mark deleted rowsNo change in indexNo change in dictionary49数据更新UpdateBase FileSegmentCarbonData TableDelete DeltaInsert DeltaBitmap file for delete delta and regular data file for inserted deltaNew index added for inserted deltaDictionary append if neededCompac

33、tionCarbon FileCarbon FileCarbon FileDictionaryIndexData FileSegment 2Segment 1Segment 1.1CarbonData TableDelta FileDelta File50性能测试51TPC-H benchmark 测试 (500GB)真实场景测试（十亿级数据)可扩展性测试 (100个节点，万亿级数据，103TB)存储层Parquet:分区列：time column (c1)CarbonData:多维索引：c1c10计算层Spark 2.152测试用例TPC-H: 查询速度0500100025002000150

34、03000500GB，TPC-H查询响应时间 （秒）parquetcarbon531.5X2.0X6.0X 2.5X1.5X1.5X2.3X为什么Query12快了6倍？/* TPC_H Query 12 - Shipping Modes and Order Priority */ SELECT L_SHIPMODE,SUM(CASE WHEN O_ORDERPRIORITY = 1-URGENT OR O_ORDERPRIORITY = 2- HIGH THEN 1 ELSE 0 END) AS HIGH_LINE_COUNT,SUM(CASE WHEN O_ORDERPRIORITY 1-

35、URGENT AND O_ORDERPRIORITY 2-HIGH THEN 1 ELSE 0 END ) AS LOW_LINE_COUNT FROM ORDERS, LINEITEMWHERE O_ORDERKEY = L_ORDERKEY AND L_SHIPMODE IN (MAIL,SHIP)AND L_COMMITDATE L_RECEIPTDATE AND L_SHIPDATE = 1994-01-01AND L_RECEIPTDATE dateadd(mm, 1, cast(1995-09-01 as date) GROUP BY L_SHIPMODEORDER BY L_SH

36、IPMODEParquet资源使用(nmon)Carbon资源使用(nmon)54TPC-H: 入库速度和压缩率52416050403020100parquetcarbonLoading Thoughput (MB/Sec/Node)553.072.843.503.002.502.001.501.000.500.00parquetcarbonCompression Ratio (higher is better)Filter响应时间(sec)Task数Queryc1c2c3c4c5c6c10ParquetCarbonDataParquetCarbonDataQ16.41.3555Q2651.3

37、8045Q3715.28049Q5644.78049Q4672.7804161Q6623.7804161Q76321.85804588Q86911.2804645过滤查询，c1到c10组合条件过滤56Carbon通过利用索引，减少Task任务下发，同时单Task内减少磁盘IO过滤测试（真实场景）结果l 过滤查询：5倍数据量，时间增加1倍。l 汇总分析：有效利用计算资源，可线性扩展 数据：2000亿到1万亿数据 (中国某省份半年 数据)集群: 70 nodes, 1120 cores查询:Q1: filter (c1c4), select * Q2: filter (c10), select * Q3: full scan aggregate0.00%100.00%200.00%300.00%400.00%500.00%200B800B1000BResponse Time (%