大数据处理引擎

1、传统的海量数据分析方案专有硬件价格昂贵可扩展能力差针对OLTP进行了优化12006年6月2007年1月2007年8月05001,0002,0003,0004,0002 Billion交易/天3 Billion交易/天4.9Billion交易/天日交易量量实例：纽纽约证券券交易所所（NYSE）5,000数据量超超过200TB基于Oracle进行行了2年年半的努努力以失失败而告告终数据量的的增长相相当快花费1000万美元构构建的数数据仓库库4个月里容容量和性性能就无无法再满满足要求求2020-02-11数据量增增长的速速度远远远超过了传统DBMS的处理能能力ApacheHadoop项目2004年D

2、ong Cutting参考Google的论文开开始实现现Hadoop2006年Yahoo年雇佣Dong Cutting组队专职职开发2006年2月，ApacheHadoop开源项目目正式启启动。ApacheHadoopHDFS设计1个NameNode+若干个DataNode将文件划划分为若若干块（Chunk）存储于于各DataNode节点上每个块固固定大小小（64M）通过冗余余来提高高可靠性性每个数据据块（Chunk）至少在在3个DataNode上冗余通过单个个NameNode来协调数数据访问问、元数数据存储储结构简单单，容易易保持元元数据一一致性HDFS体系结构构NamenodeB动态复制制

3、机架1机架2客户端（写文件件）BlocksDatanodesDatanodes客户端(读文件）查找元数数据获得文件件存储位位置元数据(文件名,存储位置置)MapReduce-“分而治之之”Inputsplitshuffleoutput交换聚集查询询（SUM、AVG等)MapMapMapMapMapReduceP1P2P3P4P5D1D2D3D4D5最终查询询结果MapReduce用户自定定义Map、ReduceWorkerWorkerMasterWorkerWorkerWorkerforkforkfork分配map分配reduce读取写入本地文件系统OutputFile 0OutputFile

4、 1写入GFSChunk 0Chunk 1Chunk 2存储于GFS的输入数据MapReduce执行流程程提交任务调度mapmapreducereduceMapReduce执行流程程GFS数据块 1数据块 2mapmapreducereduce读输入文文件MapReduce执行流程程mapmapreducereduceLocalFSLocalFS通知完成文件位置MapReduce执行流程程mapmapreducereduce本地文件件系统本地文件件系统HTTP GETMapReduce执行流程程reducereduceGFS写入最终查询结果MapReduce容错用户自定定义Map、Reduce

5、WorkerWorkerMasterWorkerWorkerforkforkfork分配map分配reduce读取Worker写入本地地文件系统统OutputFile 0OutputFile 1写入GFSChunk 0Chunk 1Chunk 2输入数据据WorkerMapReduce容忍节点点异构用户自定定义Map、ReduceWorkerWorkerMasterWorkerWorkerforkforkfork分配map分配reduce读取Worker写入本地地文件系统统OutputFile 0OutputFile 1写入GFSSplit 0Split 1Split 2输入数据据Worker

6、运行速度度过慢MapReduce调优YahooPig& FaceBook Hive集群HadoopMap-ReducePig、HIveSQL自动重写写和优化用户HIVE体系结构构HDFSHive CLIDDLQueriesMapReduceMetaStoreThriftAPISerDeThriftJSON.ExecutionHive QLParserPlannerMgmt.WebUI数据模型型Tables基本数据据类型(int,float,boolean)组合数据据类型: List/Map(关联数组组)PartitionsBucketsCREATETABLEsales( id INT,item

7、sARRAYSTRUCT)PARITIONED BY (dsSTRING)CLUSTEREDBY(id)INTO 32 BUCKETS;SELECTidFROMsalesTABLESAMPLE(BUCKET 1OUT OF 32)20Hive中的数据据共享iter.1iter.2.InputHDFSreadHDFSwriteHDFSreadHDFSwriteInputquery 1query 2query 3result 1result 2result 3. . .HDFSread数据复制制、串行行化造成成大量网网络和磁磁盘IO,导致查询询效率不不高Hive优点可靠、稳稳定支持JDBC、ODB

8、C接口动态可扩扩展、具具有容错错能力UDF、自定义义数据格格式Hive缺点data shuffle时网络瓶瓶颈，Reduce要等Map结束才能能开始，不能高高效利用用网络带带宽一般一个个SQL都会解析析成多个个MRjob，Hadoop每次Job输出都直直接写HDFS，性能差差每次执行行Job都要启动动Task，花费很很多时间间，无法法做到实实时iter.1iter.2.InputSpark分布式内内存Inputquery1query2query3.一次性导导入加载至内内存后，相比网网络和磁磁盘IO快10-100倍Spark容错维护数据据的计算算过程，当计算算结果丢丢失时，重新计计算messag

9、es= 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()Block1Block2Block

10、3WorkerWorkerWorkerDrivercachedMsgs.filter(_.contains(“foo”).countcachedMsgs.filter(_.contains(“bar”).count. .tasksresultsCache1Cache2Cache3Base RDDTransformedRDDActionSpark支持的操操作mapfiltergroupBysortjoinleftOuterJoinrightOuterJoinreducecountreduceByKeygroupByKeyfirstunioncrosssamplecogrouptakepartit

11、ionBypipesave.Spark优点轻：Spark1.0核心代码码3万行，Hadoop1.09万行，2.022万行。快：Spark对小数据据集能达达到亚秒秒级的延延迟这对于HadoopMapReduce是无法想想象的（由于“心跳”间隔机机制，仅仅任务启启动就有有数秒的的延迟）。就大数据据集而言言，对典典型的迭迭代机器器学习、图计算算等应用用，Spark版本比基基于MapReduce、Hive和Pregel的实现快快上十倍倍到百倍倍。其中中内存计计算、数数据本地地性（locality）和传输输优化、调度优优化等该该居首功功。灵：Spark提供了不不同层面面的灵活活性。在实现层层，可更更换的

12、集集群调度度器、序序列化库库；在原语（Primitive）层，它它允许扩扩展新的的数据算算子、新新的数据据源、新新的language（Java和Python）；在范式（Paradigm）层，Spark支持内存存计算、多迭代代批量处处理、即即时查询询、流处处理和图图计算等等多种范范式。巧：巧在在借势和和借力。Spark借Hadoop之势，与与Hadoop无缝结合合；接着着SparkSQL借了Hive的势；Hive体系结构构Meta storeHDFSClientDriverSQLParserQueryOptimizerPhysicalPlanExecutionCLIJDBCMapReduceSh

13、ark体系结构构Meta storeHDFSClientDriverSQLParserPhysicalPlanExecutionCLIJDBCSparkCacheMgr.QueryOptimizerStinger/TezLocalmodeSparkBagel(Pregel on Spark)Shark(HiveonSpark)StreamingSparkEC2ApacheMesosYARNTez将Map和Reduce两个操作作进一步步拆分Map被拆分成成Input、Processor、Sort、Merge和OutputReduce被拆分成成Input、Shuffle、Sort、Merge、Pr

14、ocessor和Output分解后的的元操作作可以任任意灵活活组合，产生新新的操作作，经过过一些控控制程序序组装后后，可形形成一个个大的DAG作业Stinger体系结构构Meta storeHDFSClientDriverSQLParserPhysicalPlanExecutionCLIJDBCTezCacheMgr.QueryOptimizerImpala大规模并并行数据据库引擎擎(MPP)由Cloudera开发HDFSMapReduceHivePigImpalaImpala体系结构构Impala支持的数数据格式式RCFileParquet (dremel格式)CSVAVROSequence

15、File大数据引引擎测试试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）测试环境境测试环境境（续）硬件品牌：ThinkServerRD630CPU：Xeon E5-2620（双核2.00GHz）内存：8G（其中slave2和slave4为32G）硬盘：6T（其中slave2和slave4为12T）实际总共共可用空空间为44

16、.03T软件操作系统统：CentOS6.4（64bit）文件系统统：ext4测试数据据集RankingsUserVisitsLists websites and their page rank Stores server logs for each web page pageURL VARCHAR(300)sourceIP VARCHAR(116)pageRank INTdestURL VARCHAR(100)avgDuration INTvisitDate DATEadRevenue FLOATuserAgent VARCHAR(256)countryCode CHAR(3)language

17、Code CHAR(6)searchWord VARCHAR(32)duration INT测试查询询Query1：扫描查查询SELECTpageURL,pageRankFROMrankingsWHEREpageRank XQuery2：聚合查查询SELECTSUBSTR(sourceIP,1,X),SUM(adRevenue)FROMuservisits GROUP BY SUBSTR(sourceIP,1,X)Query3：关联查查询SELECTsourceIP, totalRevenue,avgPageRankFROM( SELECTsourceIP,AVG(pageRank) as a

18、vgPageRank,SUM(adRevenue)astotalRevenueFROM Rankings AS R, UserVisitsASUVWHERER.pageURL=UV.destURLANDUV.visitDate BETWEENDate(1980-01-01) ANDDate(X)GROUPBYUV.sourceIP)ORDERBYtotalRevenueDESCLIMIT1测试查询询（续）查询代号对应的SQL语句1aSELECT pageURL, pageRank FROM rankings WHERE pageRank 10001bSELECT pageURL, pageRa

19、nk FROM rankings WHERE pageRank 1001cSELECT pageURL, pageRank FROM rankings WHERE pageRank 102aSELECT SUBSTR(sourceIP, 1, 8), SUM(adRevenue) FROM uservisits GROUP BY SUBSTR(sourceIP, 1, 8)2bSELECT SUBSTR(sourceIP, 1, 10), SUM(adRevenue) FROM uservisits GROUP BY SUBSTR(sourceIP, 1, 10)2cSELECT SUBSTR(sourceIP, 1, 12), SUM(adRevenue) FROM uservisits GROUP BY SUBSTR(sourceIP, 1, 12)3aSELECT sourceIP, sum(adRevenue) as totalRevenue, avg(pageRank) as pageRank FROM rankings R