大数据处理引擎

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

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

3、件）查找元数据获得文件存储位置元数据(文件名,存储位置)MapReduce-“分而治之”Input split shuffle output 交换聚集查询（SUM、AVG等)MapMapMapMapMapReduceP1P2P3P4P5D1D2D3D4D5最终查询结果MapReduce用户自定义Map、ReduceWorkerWorkerMasterWorkerWorkerWorkerforkforkfork分配map分配reduce读取写入本地文件系统OutputFile 0OutputFile 1写入GFSChunk 0Chunk 1Chunk 2存储于GFS的输入数据MapReduce执

4、行流程提交任务调度mapmapreducereduceMapReduce执行流程GFS数据块 1数据块 2mapmapreducereduce读输入文件MapReduce执行流程mapmapreducereduceLocal FSLocal FS通知完成文件位置MapReduce执行流程mapmapreducereduce本地文件系统本地文件系统HTTP GETMapReduce执行流程reducereduceGFS写入最终查询结果MapReduce容错用户自定义Map、ReduceWorkerWorkerMasterWorkerWorkerforkforkfork分配map分配reduce读

5、取Worker写入本地文件系统OutputFile 0OutputFile 1写入GFSChunk 0Chunk 1Chunk 2输入数据WorkerMapReduce容忍节点异构用户自定义Map、ReduceWorkerWorkerMasterWorkerWorkerforkforkfork分配map分配reduce读取Worker写入本地文件系统OutputFile 0OutputFile 1写入GFSSplit 0Split 1Split 2输入数据Worker运行速度过慢MapReduce调优Yahoo Pig& FaceBook Hive集群Hadoop Map-ReducePig、

6、HIveSQL自动重写和优化用户HIVE体系结构HDFSHive CLIDDLQueriesMap ReduceMetaStoreThrift APISerDeThriftJSON.ExecutionHive QLParserPlannerMgmt. Web UI数据模型Tables基本数据类型 (int, float, boolean)组合数据类型: List / Map (关联数组)PartitionsBucketsCREATE TABLE sales( id INT, items ARRAYSTRUCT) PARITIONED BY (ds STRING)CLUSTERED BY (id

7、) INTO 32 BUCKETS;SELECT id FROM sales TABLESAMPLE (BUCKET 1 OUT OF 32)20Hive中的数据共享iter. 1iter. 2. . .InputHDFSreadHDFSwriteHDFSreadHDFSwriteInputquery 1query 2query 3result 1result 2result 3. . .HDFSread数据复制、串行化造成大量网络和磁盘IO,导致查询效率不高Hive优点可靠、稳定支持JDBC、ODBC接口动态可扩展、具有容错能力UDF、自定义数据格式Hive缺点data shuffle时网络

8、瓶颈，Reduce要等Map结束才能开始，不能高效利用网络带宽一般一个SQL都会解析成多个MR job，Hadoop每次Job输出都直接写HDFS，性能差每次执行Job都要启动Task，花费很多时间，无法做到实时iter. 1iter. 2. . .InputSpark分布式内存Inputquery 1query 2query 3. . .一次性导入加载至内存后，相比网络和磁盘IO快10-100倍Spark容错维护数据的计算过程，当计算结果丢失时，重新计算messages = textFile(.).filter(_.contains(“error”) .map(_.split(t)(2) H

9、adoopRDDpath = hdfs:/FilteredRDDfunc = _.contains(.)MappedRDDfunc = _.split()海量日志分析示例lines = spark.textFile(“hdfs:/.”)errors = lines.filter(_.startsWith(“ERROR”)messages = errors.map(_.split(t)(2)cachedMsgs = messages.cache()Block 1Block 2Block 3WorkerWorkerWorkerDrivercachedMsgs.filter(_.contains(“

10、foo”).countcachedMsgs.filter(_.contains(“bar”).count. . .tasksresultsCache 1Cache 2Cache 3Base RDDTransformed RDDActionSpark支持的操作mapfiltergroupBysortjoinleftOuterJoinrightOuterJoinreducecountreduceByKeygroupByKeyfirstunioncrosssamplecogrouptakepartitionBypipesave.Spark优点轻：Spark1.0核心代码3万行，Hadoop1.0 9

11、万行，2.0 22万行。快：Spark对小数据集能达到亚秒级的延迟这对于Hadoop MapReduce是无法想象的（由于“心跳”间隔机制，仅任务启动就有数秒的延迟）。就大数据集而言，对典型的迭代机器学习、图计算等应用，Spark版本比基于MapReduce、Hive和Pregel的实现快上十倍到百倍。其中内存计算、数据本地性 （locality）和传输优化、调度优化等该居首功。灵：Spark提供了不同层面的灵活性。在实现层，可更换的集群调度器、序列化库；在原语（Primitive）层，它允许扩展新的数据算子、新的数据源、新的language（Java和Python）；在范式（Paradigm

12、）层，Spark支持内存计算、多迭代批量处理、即时查询、流处理和图计算等多种范式。巧：巧在借势和借力。Spark借Hadoop之势，与Hadoop无缝结合；接着Spark SQL借了Hive的势；Hive体系结构Meta storeHDFS ClientDriverSQL ParserQuery OptimizerPhysical PlanExecutionCLIJDBCMapReduceShark体系结构Meta storeHDFS ClientDriverSQL ParserPhysical PlanExecutionCLIJDBCSparkCache Mgr.Query Optimize

13、rStinger/TezLocal modeSparkBagel(Pregel on Spark)Shark(Hive on Spark)Streaming SparkEC2Apache MesosYARNTez将Map和Reduce两个操作进一步拆分Map被拆分成Input、Processor、Sort、Merge和OutputReduce被拆分成Input、Shuffle、Sort、Merge、Processor和Output分解后的元操作可以任意灵活组合，产生新的操作，经过一些控制程序组装后，可形成一个大的DAG作业Stinger体系结构Meta storeHDFS ClientDriv

14、erSQL ParserPhysical PlanExecutionCLIJDBCTezCache Mgr.Query OptimizerImpala大规模并行数据库引擎(MPP)由 Cloudera开发HDFSMap ReduceHivePigImpalaImpala体系结构Impala支持的数据格式RCFileParquet (dremel格式)CSVAVROSequence File大数据引擎测试Hive：基于Hadoop的数据仓库系统。（v0.12）Shark：与Hive兼容的SQL引擎，运行在Spark计算框架之上。（v0.8.1）Impala：与Hive兼容的SQL引擎，并自带类似

15、MPP执行引擎。（v1.2.3）Stinger/Tez：Tez是新一代的Hadoop执行引擎。（v0.2.0）测试环境 测试环境（续）硬件品牌：ThinkServer RD630CPU：Xeon E5-2620（双核 2.00GHz）内存：8G（其中slave2和slave4为32G）硬盘：6T（其中slave2和slave4为12T）实际总共可用空间为44.03T软件操作系统：CentOS6.4（64bit）文件系统：ext4测试数据集RankingsUserVisitsLists websites and their page rank Stores server logs for eac

16、h web page pageURL VARCHAR(300)sourceIP VARCHAR(116)pageRank INTdestURL VARCHAR(100)avgDuration INTvisitDate DATEadRevenue FLOATuserAgent VARCHAR(256)countryCode CHAR(3)languageCode CHAR(6)searchWord VARCHAR(32)duration INT测试查询Query1：扫描查询SELECT pageURL, pageRank FROM rankings WHERE pageRank XQuery2：

17、聚合查询SELECT SUBSTR(sourceIP, 1, X), SUM(adRevenue) FROM uservisits GROUP BY SUBSTR(sourceIP, 1, X)Query3：关联查询SELECT sourceIP, totalRevenue, avgPageRankFROM ( SELECT sourceIP, AVG(pageRank) as avgPageRank, SUM(adRevenue) as totalRevenue FROM Rankings AS R, UserVisits AS UV WHERE R.pageURL = UV.destURL

18、 AND UV.visitDate BETWEEN Date(1980-01-01) AND Date(X) GROUP BY UV.sourceIP) ORDER BY totalRevenue DESC LIMIT 1测试查询（续）查询代号对应的SQL语句1aSELECT pageURL, pageRank FROM rankings WHERE pageRank 10001bSELECT pageURL, pageRank FROM rankings WHERE pageRank 1001cSELECT pageURL, pageRank FROM rankings WHERE page

19、Rank 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 JOIN (