大数据处理引擎

传统的海量数据分析方案专有硬件价格昂贵可扩展能力差针对OLTP进行了优化12006年6月2007年1月2007年8月05001,0002,0003,0004,0002Billion交易/天3Billion交易/天4.9Billion交易/天日交易量实例：纽约证券交易所（NYSE）5,000•数据量超过200TB•基于Oracle进行了2年半的努力以失败而告终•数据量的增长相当快•花费1000万美元构建的数据仓库4个月里容量和性能就无法再满足要求2019/8/1数据量增长的速度远远超过了传统DBMS的处理能力ApacheHadoop项目•2004年DongCutting参考Google的论文开始实现Hadoop•2006年Yahoo年雇佣DongCutting组队专职开发•2006年2月，ApacheHadoop开源项目正式启动。ApacheHadoopHDFS设计•1个NameNode+若干个DataNode•将文件划分为若干块（Chunk）存储于各DataNode节点上–每个块固定大小（64M）•通过冗余来提高可靠性–每个数据块（Chunk）至少在3个DataNode上冗余•通过单个NameNode来协调数据访问、元数据存储–结构简单，容易保持元数据一致性HDFS体系结构NamenodeB动态复制机架1机架2客户端（写文件）BlocksDatanodesDatanodes客户端(读文件）查找元数据获得文件存储位置元数据(文件名,存储位置)MapReduce-“分而治之”Inputsplitshuffleoutput交换聚集查询（SUM、AVG等)MapMapMapMapMapReduceP1P2P3P4P5D1D2D3D4D5最终查询结果MapReduce用户自定义Map、ReduceWorkerWorkerMasterWorkerWorkerWorkerforkforkfork分配map分配reduce读取写入本地文件系统OutputFile0OutputFile1写入GFSChunk0Chunk1Chunk2存储于GFS的输入数据MapReduce执行流程提交任务调度mapmapreducereduceMapReduce执行流程GFS数据块1数据块2mapmapreducereduce读输入文件MapReduce执行流程mapmapreducereduceLocalFSLocalFS通知完成文件位置MapReduce执行流程mapmapreducereduce本地文件系统本地文件系统HTTPGETMapReduce执行流程reducereduceGFS写入最终查询结果MapReduce容错用户自定义Map、ReduceWorkerWorkerMasterWorkerWorkerforkforkfork分配map分配reduce读取Worker写入本地文件系统OutputFile0OutputFile1写入GFSChunk0Chunk1Chunk2输入数据WorkerMapReduce容忍节点异构用户自定义Map、ReduceWorkerWorkerMasterWorkerWorkerforkforkfork分配map分配reduce读取Worker写入本地文件系统OutputFile0OutputFile1写入GFSSplit0Split1Split2输入数据Worker运行速度过慢MapReduce调优YahooPig&FaceBookHive集群HadoopMap-ReducePig、HIveSQL自动重写和优化用户HIVE体系结构HDFSHiveCLIDDLQueriesMapReduceMetaStoreThriftAPISerDeThriftJSON..ExecutionHiveQLParserPlannerMgmt.WebUI数据模型Tables基本数据类型(int,float,boolean)组合数据类型:List/Map(关联数组)PartitionsBucketsCREATETABLEsales(idINT,itemsARRAYSTRUCTid:INT,name:STRING)PARITIONEDBY(dsSTRING)CLUSTEREDBY(id)INTO32BUCKETS;SELECTidFROMsalesTABLESAMPLE(BUCKET1OUTOF32)20Hive中的数据共享iter.1iter.2...InputHDFSreadHDFSwriteHDFSreadHDFSwriteInputquery1query2query3result1result2result3...HDFSread数据复制、串行化造成大量网络和磁盘IO,导致查询效率不高Hive优点•可靠、稳定•支持JDBC、ODBC接口•动态可扩展、具有容错能力•UDF、自定义数据格式Hive缺点•datashuffle时网络瓶颈，Reduce要等Map结束才能开始，不能高效利用网络带宽•一般一个SQL都会解析成多个MRjob，Hadoop每次Job输出都直接写HDFS，性能差•每次执行Job都要启动Task，花费很多时间，无法做到实时iter.1iter.2...InputSpark分布式内存Inputquery1query2query3...一次性导入加载至内存后，相比网络和磁盘IO快10-100倍Spark容错•维护数据的计算过程，当计算结果丢失时，重新计算messages=textFile(...).filter(_.contains(“error”)).map(_.split(‘\t’)(2))HadoopRDDpath=hdfs://…FilteredRDDfunc=_.contains(...)MappedRDDfunc=_.split(…)海量日志分析示例lines=spark.textFile(“hdfs://...”)errors=lines.filter(_.startsWith(“ERROR”))messages=errors.map(_.split(‘\t’)(2))cachedMsgs=messages.cache()Block1Block2Block3WorkerWorkerWorkerDrivercachedMsgs.filter(_.contains(“foo”)).countcachedMsgs.filter(_.contains(“bar”)).count...tasksresultsCache1Cache2Cache3BaseRDDTransformedRDDActionSpark支持的操作•map•filter•groupBy•sort•join•leftOuterJoin•rightOuterJoin•reduce•count•reduceByKey•groupByKey•first•union•crosssamplecogrouptakepartitionBypipesave...Spark优点•轻：Spark1.0核心代码3万行，Hadoop1.09万行，2.022万行。•快：Spark对小数据集能达到亚秒级的延迟这对于HadoopMapReduce是无法想象的（由于“心跳”间隔机制，仅任务启动就有数秒的延迟）。就大数据集而言，对典型的迭代机器学习、图计算等应用，Spark版本比基于MapReduce、Hive和Pregel的实现快上十倍到百倍。其中内存计算、数据本地性（locality）和传输优化、调度优化等该居首功。•灵：Spark提供了不同层面的灵活性。在实现层，可更换的集群调度器、序列化库；在原语（Primitive）层，它允许扩展新的数据算子、新的数据源、新的language（Java和Python）；在范式（Paradigm）层，Spark支持内存计算、多迭代批量处理、即时查询、流处理和图计算等多种范式。•巧：巧在借势和借力。Spark借Hadoop之势，与Hadoop无缝结合；接着SparkSQL借了Hive的势；Hive体系结构MetastoreHDFSClientDriverSQLParserQueryOptimizerPhysicalPlanExecutionCLIJDBCMapReduceShark体系结构MetastoreHDFSClientDriverSQLParserPhysicalPlanExecutionCLIJDBCSparkCacheMgr.QueryOptimizerStinger/TezLocalmodeSparkBagel(PregelonSpark)Shark(HiveonSpark)…StreamingSparkEC2ApacheMesosYARNTez•将Map和Reduce两个操作进一步拆分–Map被拆分成Input、Processor、Sort、Merge和Output–Reduce被拆分成Input、Shuffle、Sort、Merge、Processor和Output•分解后的元操作可以任意灵活组合，产生新的操作，经过一些控制程序组装后，可形成一个大的DAG作业Stinger体系结构MetastoreHDFSClientDriverSQLParserPhysicalPlanExecutionCLIJDBCTezCacheMgr.QueryOptimizerImpala►大规模并行数据库引擎(MPP)►由Cloudera开发HDFSMapReduceHivePigImpalaImpala体系结构Impala支持的数据格式►RCFile►Parquet(dremel格式)►CSV►AVRO►SequenceFile大数据引擎测试•Hive：基于Hadoop的数据仓库系统。（v0.12）•Shark：与Hive兼容的SQL引擎，运行在Spark计算框架之上。（v0.8.1）•Impala：与Hive兼容的SQL引擎，并自带类似MPP执行引擎。（v1.2.3）•Stinger/Tez：Tez是新一代的Hadoop执行引擎。（v0.2.0）测试环境测试环境（续）•硬件–品牌：ThinkServerRD630–CPU：XeonE5-2620（双核2.00GHz）–内存：8G（其中slave2和slave4为32G）–硬盘：6T（其中slave2和slave4为12T）实际总共可用空间为44.03T•软件–操作系统：CentOS6.4（64bit）–文件系统：ext4测试数据集RankingsUserVisitsListswebsitesandtheirpagerankStoresserverlogsforeachwebpagepageURLVARCHAR(300)sourceIPVARCHAR(116)pageRankINTdestURLVARCHAR(100)avgDurationINTvisitDateDATEadRevenueFLOATuserAgentVARCHAR(256)countryCodeCHAR(3)languageCodeCHAR(6)searchWordVARCHAR(32)durationINT测试查询•Query1：扫描查询–SELECTpageURL,pageRankFROMrankingsWHEREpageRankX•Query2：聚合查询–SELECTSUBSTR(sourceIP,1,X),SUM(adRevenue)FROMuservisitsGROUPBYSUBSTR(sourceIP,1,X)•Query3：关联查询–SELECTsourceIP,totalRevenue,avgPageRank–FROM–(SELECTsourceIP,–AVG(pageRank)asavgPageRank,–SUM(adRevenue)astotalRevenue–FROMRankingsASR,UserVisitsASUV–WHERER.pageURL=UV.de