CarbonData大數(shù)據(jù)交互式分析實(shí)踐

1、CarbonData: 大數(shù)據(jù)交互式分析實(shí)踐Agenda2為什么需要CarbonDataCarbonData介紹性能測(cè)試應(yīng)用案例未來(lái)計(jì)劃企業(yè)中包含多種數(shù)據(jù)應(yīng)用，從商業(yè)智能、批 處理到機(jī)器學(xué)習(xí)Big TableEx. CDR, transaction, Web log,Small tableSmall tableUnstructured datadataReport & DashboardOLAP & Ad-hocBatch processingMachine learningRealtime Analytics3來(lái)自數(shù)據(jù)的挑戰(zhàn)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 (萬(wàn)億）百億級(jí)數(shù)據(jù)量多維度細(xì)粒度來(lái)自應(yīng)用的挑戰(zhàn)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企業(yè)應(yīng)用集成5靈活查詢 無(wú)固定模式How to choose storage?6如何構(gòu)建數(shù)據(jù)平臺(tái)？選擇1: NoSQL DatabaseKey-Value store: low latency, 5ms只能通過(guò)Key訪問(wèn)，一鍵一值適合實(shí)時(shí)應(yīng)用對(duì)接，不適合分析型應(yīng)用7選擇2：Parallel databaseParallel

4、 scan + Fast computeQuestionable scalability and fault-toleranceCluster size 2.2Spark 1.5, 1.6, 2.1接口SQLDataFrame API支持操作:查詢：支持SparkSQL優(yōu)化器數(shù)據(jù)管理：批量入庫(kù)、更新、刪除、合并（Compaction）、增刪列Carbon FileCarbon FileCarbon FileCarbonData大數(shù)據(jù)生態(tài)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;使用方式：入庫(kù)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表格查詢數(shù)據(jù)管理22CarbonData介紹CarbonData文件CarbonData表格查詢數(shù)據(jù)管理23Blocklet: 文件內(nèi)的數(shù)據(jù)塊Data are sorted along MDK (multi-dimensional keys)Clustered data enabling efficient filtering and scanColumn chunk: Blocklet內(nèi)的列數(shù)據(jù)一次IO單元內(nèi)部分為多個(gè)Page，解壓?jiǎn)卧獢?shù)據(jù)和索引信息Header：Ver

9、sion，SchemaFooter： Blocklet Index & statistics內(nèi)置索引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數(shù)據(jù)和索引合一存儲(chǔ)，數(shù)據(jù)即索引多維索引 (Multi-Dimensional Key)全局字典編碼Blocklet (Columnar view)Sorted MDK IndexC1C2C3C4C5C6C71111114211432111134434462211132541547021121454

13、5588711121567556181113175705101811328561552451133652974912411144115321243952550398索引建立過(guò)程介紹25索引建立過(guò)程介紹（二）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利用兩級(jí)索引架構(gòu)減少Spark Task數(shù)和磁盤IOSpark DriverExecutorDataFooterData

16、FooterCarbon FileCarbon FileCarbon FileDataFooterBlocklet IndexBlock IndexExecutorBlocklet IndexCatalyst27第一級(jí)：文件級(jí)索引用于過(guò)濾文件（HDFS Block）， 避免掃描不必要的文件，減少多 達(dá)95%的Spark Task第二級(jí)：Blocklet索引用于過(guò)濾文件內(nèi)部的Blocklet， 避免掃描不必要的Blocklet，減 少磁盤IO支持嵌套數(shù)據(jù)類型：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表格查詢數(shù)據(jù)管理29CarbonData表格組成CarbonData FileMetadata:Global Dictionary, SchemaIndex & StatisticsImmutable Data FileData sourceLoad in batchSegmentCarbonData TableDelta FileDelta File for Update/DeleteIndex File30C

19、arbonData表格組成：多次入庫(kù)Carbon FileDictionary (append) Schema (Overwrite)Index (new tree)Data File (new folder)Data sourceLoad in batchSegment 2Segment 1CarbonData TableDelta FileCarbon FileDelta File31CarbonData表格物理存儲(chǔ)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表格查詢數(shù)據(jù)管理34文件級(jí)對(duì)接：MapReduceCarbo n FileCarbon FileCarbon FileHDFSIndexInputFormat/OutputFormatSegment 3Segment 2Segment 135SQL引擎對(duì)接：表格Carbo n FileCarbon FileCarbon FileHDFSReadSQ

22、LIndexSegment 3Segment 2Segment 136SparkSQL深度對(duì)接Carbo n FileCarbon FileCarbon File與SparkSQL優(yōu)化器 深度集成CarbonData TableRead/Write/Update/Delete/Alter TableSQLSegment 3Segment 2Segment 137Metadata: DictionaryIndexSparkSQL Data Source對(duì)接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利用兩級(jí)索引架構(gòu)減少Spark Task數(shù)和磁盤IOSpark DriverExecutorDataFooterDataFooterCarbon FileCarbon FileCarbon FileDataFooterBlocklet IndexBlock IndexExecutorBlocklet IndexCatalyst39第一級(jí)：文件級(jí)索引用于過(guò)濾文件（HDFS Block）， 避免掃描不必要的文件，減少多 達(dá)95%的Spark Task第二級(jí)：Blocklet索引用于過(guò)濾文件內(nèi)部的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一個(gè)完整的過(guò)濾查詢過(guò)程大顆粒IO單元（Carbon V3格式）：Blocklet內(nèi)部一個(gè)Column Chunk 是一個(gè)IO單元， Blocklet按大小切分，默認(rèn)64MB，大概有100萬(wàn)行記 錄。大顆粒順序讀提升掃描性能。跳躍解碼（Carbon V3格式） ：增加數(shù)據(jù)頁(yè)P(yáng)age概念，按Page進(jìn)行過(guò)濾和解碼，減 少不必要的解碼解壓縮，提升CPU利用率向量化解碼 + S

27、park向量化處理解碼和解壓縮采用向量化處理，與Spark2.1向量化、 Codegen結(jié)合，在掃描+聚合場(chǎng)景下提升性能4X堆外內(nèi)存：處理大結(jié)果集時(shí)，解碼解壓縮過(guò)程中堆外完成， 減少GCHeaderVersionSchemaTimestamp向量化CarbonReaderColumnBatch結(jié)構(gòu)Spark向量化處理 + Whole Stage CodeGen文件內(nèi)掃描優(yōu)化，與Spark向量化處理對(duì)接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表格查詢數(shù)據(jù)管理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增量入庫(kù)：排序Key和排序范圍通過(guò)SORT_COLUMNS指定需要排序的Key。如果不指定，系統(tǒng)會(huì)使用默認(rèn)規(guī)則（查詢性能最大化）：在所有維度列上建立MDK索引排序范圍是LOCAL_SORT每次入庫(kù)時(shí)可指定三種排序范圍：NO_SORT：不排序，在Compaction時(shí)建立索引。BATCH_SORT：使用配置的內(nèi)存大小進(jìn)行“盡

30、量排序”LOCAL_SORT：在節(jié)點(diǎn)內(nèi)部排序（Merge Sort）入庫(kù)流水線介紹InputConvertIndex- BuildWrite讀取和解析輸入（CSV或RDD）字典編碼轉(zhuǎn)換， 嵌套類型拉平按照SORT_SCOPE進(jìn)行排序列式編碼、壓縮、 寫文件MapperSpark TaskCarbon Data Load Pipeline45全局字典的生成方式Pre-defined Dictionary入庫(kù)前手工生成字典LOAD DATA path INTO TABLE t OPTIONS(COLUMNDICT=nplist)nplist := column_name:path,場(chǎng)景：通過(guò)靜態(tài)維

31、表預(yù)生產(chǎn)字典Two Pass Job1：生成字典 Job2：執(zhí)行入庫(kù)流程場(chǎng)景：首次入庫(kù)Single Pass由Dictionary Server在 線生成字典Dictionary Server場(chǎng)景：非首次入庫(kù)；若維度Cardinality不高， 也可用于首次入庫(kù)46數(shù)據(jù)更新和刪除47場(chǎng)景：批量更新的事實(shí)表，如更新某些度量值，ID值緩慢變化的維表，如刷新用戶信息ACID屬性：支持更新語(yǔ)句原子性（要么成功要么失?。?，一致性（更新串行化）更新過(guò)程中不影響查詢適用于OLAP型+批量更新場(chǎng)景，不支持毫秒級(jí)實(shí)時(shí)更新，不支持OLTP類應(yīng)用刪除數(shù)據(jù)DeleteBase FileSegmentCarbonDat

32、a TableDelete Delta48Bitmap file that mark deleted rowsNo change in indexNo change in dictionary49數(shù)據(jù)更新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性能測(cè)試51TPC-H benchmark 測(cè)試 (500GB)真實(shí)場(chǎng)景測(cè)試（十億級(jí)數(shù)據(jù))可擴(kuò)展性測(cè)試 (100個(gè)節(jié)點(diǎn)，萬(wàn)億級(jí)數(shù)據(jù)，103TB)存儲(chǔ)層Parquet:分區(qū)列：time column (c1)CarbonData:多維索引：c1c10計(jì)算層Spark 2.152測(cè)試用例TPC-H: 查詢速度0500100025002000150

34、03000500GB，TPC-H查詢響應(yīng)時(shí)間 （秒）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: 入庫(kù)速度和壓縮率52416050403020100parquetcarbonLoading Thoughput (MB/Sec/Node)553.072.843.503.002.502.001.501.000.500.00parquetcarbonCompression Ratio (higher is better)Filter響應(yīng)時(shí)間(sec)Task數(shù)Queryc1c2c3c4c5c6c10ParquetCarbonDataParquetCarbonDataQ16.41.3555Q2651.3

37、8045Q3715.28049Q5644.78049Q4672.7804161Q6623.7804161Q76321.85804588Q86911.2804645過(guò)濾查詢，c1到c10組合條件過(guò)濾56Carbon通過(guò)利用索引，減少Task任務(wù)下發(fā)，同時(shí)單Task內(nèi)減少磁盤IO過(guò)濾測(cè)試（真實(shí)場(chǎng)景）結(jié)果l 過(guò)濾查詢：5倍數(shù)據(jù)量，時(shí)間增加1倍。l 匯總分析：有效利用計(jì)算資源，可線性擴(kuò)展 數(shù)據(jù)：2000億到1萬(wàn)億數(shù)據(jù) (中國(guó)某省份半年 數(shù)據(jù))集群: 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 (%