传统的海量数据分析方案课件

传统的海量数据分析方案专有硬件价格昂贵可扩展能力差针对OLTP进行了优化1传统的海量数据分析方案专有硬件可扩展能力差12006年6月2007年1月2007年8月05001,0002,0003,0004,0002Billion交易/天3Billion交易/天4.9Billion交易/天日交易量实例：纽约证券交易所（NYSE）5,000数据量超过200TB基于Oracle进行了2年半的努力以失败而告终数据量的增长相当快花费1000万美元构建的数据仓库4个月里容量和性能就无法再满足要求2022/12/16数据量增长的速度远远超过了传统DBMS的处理能力2006年6月2007年1月2007年8月05001,000ApacheHadoop项目2004年DongCutting参考Google的论文开始实现Hadoop2006年Yahoo年雇佣DongCutting组队专职开发2006年2月，ApacheHadoop开源项目正式启动。ApacheHadoop项目2004年DongCuttApacheHadoopApacheHadoopHDFS设计1个NameNode+若干个DataNode将文件划分为若干块（Chunk）存储于各DataNode节点上每个块固定大小（64M）通过冗余来提高可靠性每个数据块（Chunk）至少在3个DataNode上冗余通过单个NameNode来协调数据访问、元数据存储结构简单，容易保持元数据一致性HDFS设计1个NameNode+若干个DataNodeHDFS体系结构NamenodeB动态复制机架1机架2客户端（写文件）BlocksDatanodesDatanodes客户端(读文件）查找元数据获得文件存储位置元数据(文件名,存储位置)HDFS体系结构NamenodeB动态复制机架1机架2客户端MapReduce-“分而治之”Inputsplitshuffleoutput交换MapReduce-“分而治之”Inputsplitsh聚集查询（SUM、AVG等)MapMapMapMapMapReduceP1P2P3P4P5D1D2D3D4D5最终查询结果聚集查询（SUM、AVG等)MapMapMapMapMapRMapReduce用户自定义Map、ReduceWorkerWorkerMasterWorkerWorkerWorkerforkforkfork分配map分配reduce读取写入本地文件系统OutputFile0OutputFile1写入GFSChunk0Chunk1Chunk2存储于GFS的输入数据MapReduce用户自定义WorkerWorkerMastMapReduce执行流程提交任务调度mapmapreducereduceMapReduce执行流程提交任务调度mapmapreducMapReduce执行流程GFS数据块1数据块2mapmapreducereduce读输入文件MapReduce执行流程GFS数据块1数据块2mapmMapReduce执行流程mapmapreducereduceLocalFSLocalFS通知完成文件位置MapReduce执行流程mapmapreducereducMapReduce执行流程mapmapreducereduce本地文件系统本地文件系统HTTPGETMapReduce执行流程mapmapreducereducMapReduce执行流程reducereduceGFS写入最终查询结果MapReduce执行流程reducereduceGFS写入MapReduce容错用户自定义Map、ReduceWorkerWorkerMasterWorkerWorkerforkforkfork分配map分配reduce读取Worker写入本地文件系统OutputFile0OutputFile1写入GFSChunk0Chunk1Chunk2输入数据WorkerMapReduce容错用户自定义WorkerWorkerMaMapReduce容忍节点异构用户自定义Map、ReduceWorkerWorkerMasterWorkerWorkerforkforkfork分配map分配reduce读取Worker写入本地文件系统OutputFile0OutputFile1写入GFSSplit0Split1Split2输入数据Worker运行速度过慢MapReduce容忍节点异构用户自定义WorkerWorkMapReduce调优MapReduce调优YahooPig&FaceBookHive集群HadoopMap-ReducePig、HIveSQL自动重写和优化用户YahooPig&FaceBookHive集群HadoHIVE体系结构HDFSHiveCLIDDLQueriesMapReduceMetaStoreThriftAPISerDeThriftJSON..ExecutionHiveQLParserPlannerMgmt.WebUIHIVE体系结构HDFSHiveCLIDDLQueries数据模型Tables基本数据类型(int,float,boolean)组合数据类型:List/Map(关联数组)PartitionsBucketsCREATETABLEsales(idINT,itemsARRAY<STRUCT<id:INT,name:STRING> )PARITIONEDBY(dsSTRING)

CLUSTEREDBY(id)INTO32BUCKETS;SELECTidFROMsalesTABLESAMPLE(BUCKET1OUTOF32)20数据模型Tables20Hive中的数据共享iter.1iter.2...InputHDFS

readHDFS

writeHDFS

readHDFS

writeInputquery1query2query3result1result2result3...HDFS

read数据复制、串行化造成大量网络和磁盘IO,导致查询效率不高Hive中的数据共享iter.1iter.2..Hive优点可靠、稳定支持JDBC、ODBC接口动态可扩展、具有容错能力UDF、自定义数据格式Hive优点可靠、稳定Hive缺点datashuffle时网络瓶颈，Reduce要等Map结束才能开始，不能高效利用网络带宽一般一个SQL都会解析成多个MRjob，Hadoop每次Job输出都直接写HDFS，性能差每次执行Job都要启动Task，花费很多时间，无法做到实时Hive缺点datashuffle时网络瓶颈，Reduceiter.1iter.2...InputSpark分布式内存Inputquery1query2query3...一次性导入加载至内存后，相比网络和磁盘IO快10-100倍iter.1iter.2...InputSparSpark容错维护数据的计算过程，当计算结果丢失时，重新计算messages=textFile(...).filter(_.contains(“error”)).map(_.split(‘\t’)(2))

HadoopRDDpath=hdfs://…FilteredRDDfunc=_.contains(...)MappedRDDfunc=_.split(…)Spark容错维护数据的计算过程，当计算结果丢失时，重新计算海量日志分析示例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...tasksresultsCache1Cache2Cache3BaseRDDTransformedRDDAction海量日志分析示例lines=spark.textFileSpark支持的操作mapfiltergroupBysortjoinleftOuterJoinrightOuterJoinreducecountreduceByKeygroupByKeyfirstunioncrosssamplecogrouptakepartitionBypipesave...Spark支持的操作mapreducesampleSpark优点轻：Spark1.0核心代码3万行，Hadoop1.0

9万行，2.0

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

SQL借了Hive的势；Spark优点轻：Spark1.0核心代码3万行，HadooHive体系结构MetastoreHDFSClientDriverSQLParserQueryOptimizerPhysicalPlanExecutionCLIJDBCMapReduceHive体系结构MetastoreHDFSDriverSQShark体系结构MetastoreHDFSClientDriverSQLParserPhysicalPlanExecutionCLIJDBCSparkCacheMgr.QueryOptimizerShark体系结构MetastoreHDFSDriverSStinger/TezLocalmodeSparkBagel(PregelonSpark)Shark

(HiveonSpark)…StreamingSparkEC2ApacheMesosYARNStinger/TezLocalmodeSparkBageTez将Map和Reduce两个操作进一步拆分Map被拆分成Input、Processor、Sort、Merge和OutputReduce被拆分成Input、Shuffle、Sort、Merge、Processor和Output分解后的元操作可以任意灵活组合，产生新的操作，经过一些控制程序组装后，可形成一个大的DAG作业Tez将Map和Reduce两个操作进一步拆分Stinger体系结构MetastoreHDFSClientDriverSQLParserPhysicalPlanExecutionCLIJDBCTezCacheMgr.QueryOptimizerStinger体系结构MetastoreHDFSDriveImpala大规模并行数据库引擎(MPP)由Cloudera开发HDFSMapReduceHivePigImpalaImpala大规模并行数据库引擎(MPP)HDFSMapRImpala体系结构Impala体系结构Impala支持的数据格式RCFileParquet(dremel格式)CSVAVROSequenceFileImpala支持的数据格式RCFile大数据引擎测试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）大数据引擎测试Hive：基于Hadoop的数据仓库系统。（v测试环境

测试环境测试环境（续）硬件品牌：ThinkServerRD630CPU：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测试数据集RankingsUserVisitsListsw测试查询Query1：扫描查询SELECTpageURL,pageRankFROMrankingsWHEREpageRank>XQuery2：聚合查询SELECTSUBSTR(sourceIP,1,X),SUM(adRevenue)FROMuservisitsGROUPBYSUBSTR(sourceIP,1,X)Query3：关联查询SELECTsourceIP,totalRevenue,avgPageRankFROM(SELECTsourceIP,AVG(pageRank)asavgPageRank,SUM(adRevenue)astotalRevenueFROMRankingsASR,UserVisitsASUVWHERER.pageURL=UV.destURLANDUV.visitDateBETWEENDate(‘1980-01-01')ANDDate(‘X')GROUPBYUV.sourceIP)ORDERBYtotalRevenueDESCLIMIT1测试查询Query1：扫描查询测试查询（续）查询代号对应的SQL语句1aSELECTpageURL,pageRankFROMrankingsWHEREpageRank>10001bSELECTpageURL,pageRankFROMrankingsWHEREpageRank>1001cSELECTpageURL,pageRankFROMrankingsWHEREpageRank>102aSELECTSUBSTR(sourceIP,1,8),SUM(adRevenue)FROMuservisitsGROUPBYSUBSTR(sourceIP,1,8)2bSELECTSUBSTR(sourceIP,1,10),SUM(adRevenue)FROMuservisitsGROUPBYSUBSTR(sourceIP,1,10)2cSELECTSUBSTR(sourceIP,1,12),SUM(adRevenue)FROMuservisitsGROUPBYSUBSTR(sourceIP,1,12)3aSELECTsourceIP,sum(adRevenue)astotalRevenue,avg(pageRank)aspageRankFROM

rankingsRJOIN(SELECTsourceIP,destURL,adRevenueFROMuservisitsUVWHEREUV.visitDate>"1980-01-01"ANDUV.visitDate<"1980-04-01")NUVON(R.pageURL=NUV.destURL)GROUPBYsourceIPORDERBYtotalRevenueDESCLIMIT13bSELECTsourceIP,sum(adRevenue)astotalRevenue,avg(pageRank)aspageRankFROM

rankingsRJOIN(SELECTsourceIP,destURL,adRevenueFROMuservisitsUVWHEREUV.visitDate>"1980-01-01"ANDUV.visitDate<"1983-01-01")NUVON(R.pageURL=NUV.destURL)GROUPBYsourceIPORDERBYtotalRevenueDESC

LIMIT13cSELECTsourceIP,sum(adRevenue)astotalRevenue,avg(pageRank)aspageRankFROM

rankingsRJOIN(SELECTsourceIP,destURL,adRevenueFROMuservisitsUVWHEREUV.visitDate>"1980-01-01"ANDUV.visitDate<"2010-01-01")NUVON(R.pageURL=NUV.destURL)GROUPBYsourceIPORDERBYtotalRevenueDESC

LIMIT1测试查询（续）查询代号对应的SQL语句1aSELECTpa测试结果系统\查询1a1b1cHive38.795950.9670712.7298Tez21.657129.8249411.1725Impala14.174720.7351197.1274NHCloudDB6.11428.6905113.3763测试结果系统\查询1a1b1cHive38.795950测试结果（续1）系统\查询2a2b2cHive225.1284397.1289872.6722Tez98.3167168.9715469.5765Impala43.3130113.7477326.8514NHCloudDB42.058062.0554142.4351测试结果（续1）系统\查询2a2b2cHive225.1测试结果（续2）系统\查询3a3b3cHive997.2121821.3242027.045Tez396.528478.0121145.279Impala371.015456.9461046.453NHCloudDB341.7607398.2535872.1971测试结果（续2）系统\查询3a3b3cHive997.2演讲完毕，谢谢观看！演讲完毕，谢谢观看！传统的海量数据分析方案专有硬件价格昂贵可扩展能力差针对OLTP进行了优化47传统的海量数据分析方案专有硬件可扩展能力差12006年6月2007年1月2007年8月05001,0002,0003,0004,0002Billion交易/天3Billion交易/天4.9Billion交易/天日交易量实例：纽约证券交易所（NYSE）5,000数据量超过200TB基于Oracle进行了2年半的努力以失败而告终数据量的增长相当快花费1000万美元构建的数据仓库4个月里容量和性能就无法再满足要求2022/12/16数据量增长的速度远远超过了传统DBMS的处理能力2006年6月2007年1月2007年8月05001,000ApacheHadoop项目2004年DongCutting参考Google的论文开始实现Hadoop2006年Yahoo年雇佣DongCutting组队专职开发2006年2月，ApacheHadoop开源项目正式启动。ApacheHadoop项目2004年DongCuttApacheHadoopApacheHadoopHDFS设计1个NameNode+若干个DataNode将文件划分为若干块（Chunk）存储于各DataNode节点上每个块固定大小（64M）通过冗余来提高可靠性每个数据块（Chunk）至少在3个DataNode上冗余通过单个NameNode来协调数据访问、元数据存储结构简单，容易保持元数据一致性HDFS设计1个NameNode+若干个DataNodeHDFS体系结构NamenodeB动态复制机架1机架2客户端（写文件）BlocksDatanodesDatanodes客户端(读文件）查找元数据获得文件存储位置元数据(文件名,存储位置)HDFS体系结构NamenodeB动态复制机架1机架2客户端MapReduce-“分而治之”Inputsplitshuffleoutput交换MapReduce-“分而治之”Inputsplitsh聚集查询（SUM、AVG等)MapMapMapMapMapReduceP1P2P3P4P5D1D2D3D4D5最终查询结果聚集查询（SUM、AVG等)MapMapMapMapMapRMapReduce用户自定义Map、ReduceWorkerWorkerMasterWorkerWorkerWorkerforkforkfork分配map分配reduce读取写入本地文件系统OutputFile0OutputFile1写入GFSChunk0Chunk1Chunk2存储于GFS的输入数据MapReduce用户自定义WorkerWorkerMastMapReduce执行流程提交任务调度mapmapreducereduceMapReduce执行流程提交任务调度mapmapreducMapReduce执行流程GFS数据块1数据块2mapmapreducereduce读输入文件MapReduce执行流程GFS数据块1数据块2mapmMapReduce执行流程mapmapreducereduceLocalFSLocalFS通知完成文件位置MapReduce执行流程mapmapreducereducMapReduce执行流程mapmapreducereduce本地文件系统本地文件系统HTTPGETMapReduce执行流程mapmapreducereducMapReduce执行流程reducereduceGFS写入最终查询结果MapReduce执行流程reducereduceGFS写入MapReduce容错用户自定义Map、ReduceWorkerWorkerMasterWorkerWorkerforkforkfork分配map分配reduce读取Worker写入本地文件系统OutputFile0OutputFile1写入GFSChunk0Chunk1Chunk2输入数据WorkerMapReduce容错用户自定义WorkerWorkerMaMapReduce容忍节点异构用户自定义Map、ReduceWorkerWorkerMasterWorkerWorkerforkforkfork分配map分配reduce读取Worker写入本地文件系统OutputFile0OutputFile1写入GFSSplit0Split1Split2输入数据Worker运行速度过慢MapReduce容忍节点异构用户自定义WorkerWorkMapReduce调优MapReduce调优YahooPig&FaceBookHive集群HadoopMap-ReducePig、HIveSQL自动重写和优化用户YahooPig&FaceBookHive集群HadoHIVE体系结构HDFSHiveCLIDDLQueriesMapReduceMetaStoreThriftAPISerDeThriftJSON..ExecutionHiveQLParserPlannerMgmt.WebUIHIVE体系结构HDFSHiveCLIDDLQueries数据模型Tables基本数据类型(int,float,boolean)组合数据类型:List/Map(关联数组)PartitionsBucketsCREATETABLEsales(idINT,itemsARRAY<STRUCT<id:INT,name:STRING> )PARITIONEDBY(dsSTRING)

CLUSTEREDBY(id)INTO32BUCKETS;SELECTidFROMsalesTABLESAMPLE(BUCKET1OUTOF32)66数据模型Tables20Hive中的数据共享iter.1iter.2...InputHDFS

readHDFS

writeHDFS

readHDFS

writeInputquery1query2query3result1result2result3...HDFS

read数据复制、串行化造成大量网络和磁盘IO,导致查询效率不高Hive中的数据共享iter.1iter.2..Hive优点可靠、稳定支持JDBC、ODBC接口动态可扩展、具有容错能力UDF、自定义数据格式Hive优点可靠、稳定Hive缺点datashuffle时网络瓶颈，Reduce要等Map结束才能开始，不能高效利用网络带宽一般一个SQL都会解析成多个MRjob，Hadoop每次Job输出都直接写HDFS，性能差每次执行Job都要启动Task，花费很多时间，无法做到实时Hive缺点datashuffle时网络瓶颈，Reduceiter.1iter.2...InputSpark分布式内存Inputquery1query2query3...一次性导入加载至内存后，相比网络和磁盘IO快10-100倍iter.1iter.2...InputSparSpark容错维护数据的计算过程，当计算结果丢失时，重新计算messages=textFile(...).filter(_.contains(“error”)).map(_.split(‘\t’)(2))

HadoopRDDpath=hdfs://…FilteredRDDfunc=_.contains(...)MappedRDDfunc=_.split(…)Spark容错维护数据的计算过程，当计算结果丢失时，重新计算海量日志分析示例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...tasksresultsCache1Cache2Cache3BaseRDDTransformedRDDAction海量日志分析示例lines=spark.textFileSpark支持的操作mapfiltergroupBysortjoinleftOuterJoinrightOuterJoinreducecountreduceByKeygroupByKeyfirstunioncrosssamplecogrouptakepartitionBypipesave...Spark支持的操作mapreducesampleSpark优点轻：Spark1.0核心代码3万行，Hadoop1.0

9万行，2.0

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

SQL借了Hive的势；Spark优点轻：Spark1.0核心代码3万行，HadooHive体系结构MetastoreHDFSClientDriverSQLParserQueryOptimizerPhysicalPlanExecutionCLIJDBCMapReduceHive体系结构MetastoreHDFSDriverSQShark体系结构MetastoreHDFSClientDriverSQLParserPhysicalPlanExecutionCLIJDBCSparkCacheMgr.QueryOptimizerShark体系结构MetastoreHDFSDriverSStinger/TezLocalmodeSparkBagel(PregelonSpark)Shark

(HiveonSpark)…StreamingSparkEC2ApacheMesosYARNStinger/TezLocalmodeSparkBageTez将Map和Reduce两个操作进一步拆分Map被拆分成Input、Processor、Sort、Merge和OutputReduce被拆分成Input、Shuffle、Sort、Merge、Processor和Output分解后的元操作可以任意灵活组合，产生新的操作，经过一些控制程序组装后，可形成一个大的DAG作业Tez将Map和Reduce两个操作进一步拆分Stinger体系结构MetastoreHDFSClientDriverSQLParserPhysicalPlanExecutionCLIJDBCTezCacheMgr.QueryOptimizerStinger体系结构MetastoreHDFSDriveImpala大规模并行数据库引擎(MPP)由Cloudera开发HDFSMapReduceHivePigImpalaImpala大规模并行数据库引擎(MPP)HDFSMapRImpala体系结构Impala体系结构Impala支持的数据格式RCFileParquet(dremel格式)CSVAVROSequenceFileImpala支持的数据格式RCFile大数据引擎测试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）大数据引擎测试Hive：基于Hadoop的数据仓库系统。（v测试环境

测试环境测试环境（续）硬件品牌：ThinkServerRD630CPU：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测试数据集RankingsUserVisitsListsw测试查询Query1：扫描查询SELECTpageURL,pageRankFROMrankingsWHEREpageRank>XQuery2：聚合查询SELECTSUBSTR(sourceIP,1,X),SUM(adRevenue)FROMuservisitsGROUPBYSUBSTR(sourceIP,1,X)Query3：关联查询SELECTsourceIP,totalRevenue,avgPageRankFROM(SELECTsourceIP,AVG(pageRank)asavgPageRank,SUM(adRevenue)astotalRevenueFROMRankingsASR,UserVisitsASUVWHERER.pageURL=UV.destURLANDUV.visitDateBETWEENDate(‘1980-01-01')ANDDate(‘X')GROUPBYUV.sourceIP)ORDERBYtotalRevenueDESCLIMIT1测试查询Query1：扫描查询测试查询（续）查询代号对应的SQL语句1aSELECTpageURL,pageRankFROMrankingsWHEREpageRank>10001bSELECTpageURL,pageRankFROMrankingsWHEREpageRank>1001cSELECTpageURL,pageRankFROMrankingsWHEREpageRank>102aSELECTSUBSTR(sourceIP,1,8),SUM(adRevenue)FROMus