基于hadoopde大數(shù)據(jù)應(yīng)用分析(4)

1、基于基于Hadoop的大數(shù)據(jù)應(yīng)用分析的大數(shù)據(jù)應(yīng)用分析靳靳 鵬鵬合肥工業(yè)大學(xué)，管理學(xué)院信管系合肥工業(yè)大學(xué)，管理學(xué)院信管系內(nèi)容提要內(nèi)容提要 1. HADOOP體系架構(gòu) 2. 基于HADOOP的大數(shù)據(jù)產(chǎn)品分析 3.3. HADOOPHADOOP部署部署大數(shù)據(jù)對(duì)系統(tǒng)的需求大數(shù)據(jù)對(duì)系統(tǒng)的需求High performance 高并發(fā)讀寫的需求高并發(fā)、實(shí)時(shí)動(dòng)態(tài)獲取和更新數(shù)據(jù)Huge Storage 海量數(shù)據(jù)的高效率存儲(chǔ)和訪問的需求類似SNS網(wǎng)站，海量用戶信息的高效率實(shí)時(shí)存儲(chǔ)和查詢High Scalability & High Availability 高可擴(kuò)展性和高可用性的需求需要擁有快速橫向擴(kuò)展能

2、力、提供7*24小時(shí)不間斷服務(wù)商業(yè)模式驅(qū)動(dòng)應(yīng)用需求驅(qū)動(dòng)p云計(jì)算改變了IT,而大數(shù)據(jù)則改變了業(yè)務(wù)p云計(jì)算是大數(shù)據(jù)的IT基礎(chǔ)，大數(shù)據(jù)須有云計(jì)算作為基礎(chǔ)架構(gòu)，才能高效運(yùn)行p通過大數(shù)據(jù)的業(yè)務(wù)需求，為云計(jì)算的落地找到了實(shí)際應(yīng)用大數(shù)據(jù)和云計(jì)算的關(guān)系大數(shù)據(jù)和云計(jì)算的關(guān)系大數(shù)據(jù)大數(shù)據(jù)云計(jì)算云計(jì)算大數(shù)據(jù)主要應(yīng)用技術(shù)大數(shù)據(jù)主要應(yīng)用技術(shù)HadoopHadoopHadoop作為新一代的架構(gòu)和技術(shù)，因?yàn)橛欣诓⑿蟹植继幚怼按髷?shù)據(jù)”而備受重視。Apache Hadoop 是一個(gè)用java語(yǔ)言實(shí)現(xiàn)的軟件框架，在由大量計(jì)算機(jī)組成的集群中運(yùn)行海量數(shù)據(jù)的分布式計(jì)算，它可以讓應(yīng)用程序支持上千個(gè)節(jié)點(diǎn)和PB級(jí)別的數(shù)據(jù)。 Hadoop是

3、項(xiàng)目的總稱，主要是由分布式 存 儲(chǔ) （ H D F S ） 、 分 布 式 計(jì) 算（MapReduce）等組成 。優(yōu)點(diǎn)：l可擴(kuò)展：不論是存儲(chǔ)的可擴(kuò)展還是計(jì)算的可擴(kuò)展都是Hadoop的設(shè)計(jì)根本。l經(jīng)濟(jì)：框架可以運(yùn)行在任何普通的PC上。l可靠：分布式文件系統(tǒng)的備份恢復(fù)機(jī)制以及MapReduce的任務(wù)監(jiān)控保證了分布式處理的可靠性。l高效：分布式文件系統(tǒng)的高效數(shù)據(jù)交互實(shí)現(xiàn)以及MapReduce結(jié)合Local Data處理的模式，為高效處理海量的信息作了基礎(chǔ)準(zhǔn)備。Hadoop系統(tǒng)工作原理Hadoop系統(tǒng)構(gòu)架MapReduceHDFSHBasePigChuKwaHiveZooKeeperHadoopHad

4、oop體系架構(gòu)體系架構(gòu)Pig是一個(gè)基于Hadoop的大規(guī)模數(shù)據(jù)分析平臺(tái)，Pig為復(fù)雜的海量數(shù)據(jù)并行計(jì)算提供了一個(gè)簡(jiǎn)易的操作和編程接口Chukwa是基于Hadoop的集群監(jiān)控系統(tǒng)，由yahoo貢獻(xiàn)hive是基于Hadoop的一個(gè)工具，提供完整的sql查詢功能，可以將sql語(yǔ)句轉(zhuǎn)換為MapReduce任務(wù)進(jìn)行運(yùn)行ZooKeeper：高效的，可擴(kuò)展的協(xié)調(diào)系統(tǒng),存儲(chǔ)和協(xié)調(diào)關(guān)鍵共享狀態(tài)HBase是一個(gè)開源的，基于列存儲(chǔ)模型的分布式數(shù)據(jù)庫(kù)HDFS是一個(gè)分布式文件系統(tǒng)。有著高容錯(cuò)性的特點(diǎn)，并且設(shè)計(jì)用來部署在低廉的硬件上，適合那些有著超大數(shù)據(jù)集的應(yīng)用程序MapReduce是一種編程模型，用于大規(guī)模數(shù)據(jù)集（大于

5、1TB）的并行運(yùn)算MapReducelMap：任務(wù)的分解lReduce：結(jié)果的匯總兩大核心設(shè)計(jì)HDFSlNameNode：文件管理lDataNode：文件存儲(chǔ)lClient：文件獲取HadoopHadoop核心設(shè)計(jì)核心設(shè)計(jì)HDFSHDFS分布式文件系統(tǒng)分布式文件系統(tǒng)NameNode可以看作是分布式文件系統(tǒng)中的管理者，存儲(chǔ)文件系統(tǒng)的meta-data，主要負(fù)責(zé)管理文件系統(tǒng)的命名空間，集群配置信息，存儲(chǔ)塊的復(fù)制。DataNode是文件存儲(chǔ)的基本單元。它存儲(chǔ)文件塊在本地文件系統(tǒng)中，保存了文件塊的meta-data，同時(shí)周期性的發(fā)送所有存在的文件塊的報(bào)告給NameNode。 Client就是需要獲取分

6、布式文件系統(tǒng)文件的應(yīng)用程序。 HDFS是一個(gè)高度容錯(cuò)性的分布式文件系統(tǒng)，能提供高吞吐量的數(shù)據(jù)訪問，非常適合大規(guī)模數(shù)據(jù)集上的應(yīng)用。HDFSHDFS具體操作具體操作文件寫入：1. Client向NameNode發(fā)起文件寫入的請(qǐng)求2. NameNode根據(jù)文件大小和文件塊配置情況，返回給Client它所管理部分DataNode的信息。3. Client將文件劃分為多個(gè)文件塊，根據(jù)DataNode的地址信息，按順序?qū)懭氲矫恳粋€(gè)DataNode塊中。文件讀?。?. Client向NameNode發(fā)起文件讀取的請(qǐng)求2. NameNode返回文件存儲(chǔ)的DataNode的信息。3. Client讀取文件信息。

7、MapReduceMapReduce映射、化簡(jiǎn)編程模型映射、化簡(jiǎn)編程模型1. 根據(jù)輸入數(shù)據(jù)的大小和參數(shù)的設(shè)置把數(shù)據(jù)分成splits, 每個(gè)split對(duì)于一個(gè)map線程。2. Split中的數(shù)據(jù)作為Map的輸入， Map的輸出一定在Map端。3. Map的輸出到Reduce的輸入的過程(shuffle過程)： 第一階段：在map端完成內(nèi)存-排序-寫入磁盤-復(fù)制 第二階段：在reduce端完成映射到reduce端分區(qū)-合并-排序4. Reduce的輸入到Reduce的輸出 最后排好序的key/value作為Reduce的輸入，輸出不一定是在reduce端。MapReduce是一種編程模型，用于大規(guī)

8、模數(shù)據(jù)集的并行運(yùn)算。Map（映射）和Reduce（化簡(jiǎn)），采用分而治之思想，先把任務(wù)分發(fā)到集群多個(gè)節(jié)點(diǎn)上，并行計(jì)算，然后再把計(jì)算結(jié)果合并，從而得到最終計(jì)算結(jié)果。多節(jié)點(diǎn)計(jì)算，所涉及的任務(wù)調(diào)度、負(fù)載均衡、容錯(cuò)處理等，都由MapReduce框架完成，不需要編程人員關(guān)心這些內(nèi)容。 HBASEHBASE分布式分布式數(shù)據(jù)存儲(chǔ)數(shù)據(jù)存儲(chǔ)HBase Hadoop Database，是一個(gè)高可靠性、高性能、面向列、可伸縮的分布式存儲(chǔ)系統(tǒng)；HBase位于結(jié)構(gòu)化存儲(chǔ)層，HDFS為HBase提供了高可靠性的底層存儲(chǔ)支持,MapReduce為HBase提供了高性能的計(jì)算能力，Zookeeper為HBase提供了穩(wěn)定服務(wù)和

9、failover機(jī)制；Pig和Hive還為HBase提供了高層語(yǔ)言支持，使得在HBase上進(jìn)行數(shù)據(jù)統(tǒng)計(jì)處理變的簡(jiǎn)單。內(nèi)容提要內(nèi)容提要 1. HADOOP體系架構(gòu) 2. 基于HADOOP的大數(shù)據(jù)產(chǎn)品分析 3.3. HADOOPHADOOP部署部署HadoopHadoop主要開發(fā)廠商主要開發(fā)廠商大型企業(yè)和機(jī)構(gòu)在尋求解決棘手的大數(shù)據(jù)問題時(shí)，往往會(huì)使用開源軟件基礎(chǔ)架構(gòu)Hadoop的服務(wù)。由于Hadoop深受歡迎，許多公司都推出了各自版本的Hadoop，也有一些公司則圍繞Hadoop提供解決方案。 Hadoop的發(fā)行版除了社區(qū)的Apache hadoop外，cloudera，IBM，ORACLE等都提供

10、了自己的商業(yè)版本。商業(yè)版主要是提供Hadoop專業(yè)的技術(shù)支持，這對(duì)一些大型企業(yè)尤其重要。ClouderaEMCIBMOracle內(nèi)容提要內(nèi)容提要 1. HADOOP體系架構(gòu) 2. 基于HADOOP的大數(shù)據(jù)產(chǎn)品分析 3. HADOOP部署l3 Sept 2008l NEUSOFT SECRETHadoop分布式存儲(chǔ)與計(jì)算框架目錄目錄l 分布式引例l Hadoop1.0架構(gòu)l Hadoop1.0編程l Hadoop1.0弊端l Hadoop2.0架構(gòu)l Hadoop2.0編程l Hadoop2.0部署分布式引例分布式引例概述概述分布式引例分布式引例問題問題場(chǎng)景場(chǎng)景假設(shè)配置完全相同的機(jī)器M01M0X

11、，并且每臺(tái)機(jī)器都是1個(gè)雙核CPU，5G硬盤?，F(xiàn)有兩個(gè)大小都是2G的文件file01和file02第一類問題：存儲(chǔ)問題將file01和file02存入兩臺(tái)不同機(jī)器，但要求對(duì)外顯示它們存于同一硬盤空間問題不考慮，現(xiàn)有一新文件file03，大小為6G，要求存入機(jī)器后對(duì)外顯示依舊為一個(gè)完整文件第二類問題：計(jì)算問題在問題下，統(tǒng)計(jì)file01和file02這兩個(gè)文件里，每個(gè)單詞出現(xiàn)的次數(shù)第三類問題：可靠性問題假設(shè)用于解決上述問題的機(jī)器崩潰了，您如何保證數(shù)據(jù)不丟失分布式引例分布式引例常規(guī)解決方案常規(guī)解決方案存儲(chǔ)問題思路：取M01，M02，直接將file01存M01，file02存M02file02存儲(chǔ)問題存

12、儲(chǔ)問題思路：file03拆成file03-a,file03-b將file03-a存M01，file03-b存M02存儲(chǔ)問題第一類問題 存儲(chǔ)若能將 M01與M02構(gòu)建成統(tǒng)一空間？思考問題解決了？問題解決了？分布式解決方案分布式引例分布式引例常規(guī)解決方案常規(guī)解決方案計(jì)算問題思路：M02存的file02拷貝至M01，M01上執(zhí)行計(jì)算計(jì)算問題第二類問題 計(jì)算M01執(zhí)行計(jì)算解決了！并且簡(jiǎn)單實(shí)用好維護(hù)可若現(xiàn)在數(shù)據(jù)不是2G，而是2T機(jī)器不是2臺(tái)，而是200臺(tái)l 僅數(shù)據(jù)copy這一步就要花去一周時(shí)間！l 哪臺(tái)服務(wù)器硬盤有400T？問題解決了嗎？若每臺(tái)機(jī)器都能l先本地計(jì)算(Map)l接著根據(jù)規(guī)則發(fā)不同機(jī)器(sh

13、uffle)l再合并計(jì)算(Reduce)那！分布式解決方案Map-Shuffle-Reduce分布式引例分布式引例常規(guī)解決方案常規(guī)解決方案可靠性問題思路：用最穩(wěn)定最貴的機(jī)器可靠性問題第三類問題 可靠性可靠性問題硬件需提供高可靠性這點(diǎn)沒錯(cuò)火山！地震！冗余！存很多備份分布式存儲(chǔ)架構(gòu)思路分布式引例分布式引例分布式解決方案分布式解決方案第一類問題 存儲(chǔ)思路l取M01，M02，M99l構(gòu)建master/slave體系分布式集群規(guī)定lM99：不做存儲(chǔ)，統(tǒng)一管理M01，M02硬盤空間lM01，M02：存儲(chǔ)具體文件存儲(chǔ)問題由于硬盤空間大，可任意存存儲(chǔ)問題由于硬盤連在一起，可任意存file01采用分布式-存儲(chǔ)問

14、題全部解決Hadoop里Hdfs架構(gòu)思想由于采用master/slave架構(gòu)，很容易添加slave存儲(chǔ)主節(jié)點(diǎn)存儲(chǔ)主節(jié)點(diǎn)存儲(chǔ)主節(jié)點(diǎn)分布式引例分布式引例分布式解決方案分布式解決方案第二類問題 計(jì)算data cstordatacstordatacstor 結(jié)果存至DFS結(jié)果存至DFS若能這樣計(jì)算WordCount，那真的太好了存儲(chǔ)主節(jié)點(diǎn)本地計(jì)算洗牌合并再計(jì)算存結(jié)果分布式存儲(chǔ)分布式計(jì)算為此構(gòu)建分布式計(jì)算見下頁(yè)P(yáng)PTl 先進(jìn)行本機(jī)計(jì)算l 再按規(guī)則發(fā)不同機(jī)器l 進(jìn)行合并計(jì)算WordCount解題思路網(wǎng)網(wǎng) 絡(luò)絡(luò)分布式計(jì)算架構(gòu)思路分布式引例分布式引例分布式解決方案分布式解決方案第二類問題 計(jì)算思路l取M01，

15、M02，M98l構(gòu)建master/slave體系分布式集群規(guī)定lM99：不做計(jì)算，統(tǒng)一管理M01，M02內(nèi)存,CPUlM01，M02：具體計(jì)算任務(wù)計(jì)算問題本地計(jì)算洗牌合并計(jì)算由于采用master/slave架構(gòu)，很容易添加slave計(jì)算主節(jié)點(diǎn) 結(jié)果存至DFS網(wǎng)網(wǎng) 絡(luò)絡(luò)計(jì)算主節(jié)點(diǎn)分分布布式存式存儲(chǔ)儲(chǔ)DFS結(jié)果存至DFS思路Hadoop-MapReduce計(jì)算思路Hadoop分布式計(jì)算架構(gòu)兩個(gè)結(jié)論，但兩者不同：一個(gè)架構(gòu)，一個(gè)計(jì)算思路cstordatadatacstordatadatal3 Sept 2008l NEUSOFT SECRET public void map(Object key, T

16、ext value, Context context ) StringTokenizer itr = new StringTokenizer(value.toString(); while (itr.hasMoreTokens() word.set(itr.nextToken(); context.write(word, 1); public void reduce(Text key, Iterable values, Context context) int sum = 0; for (IntWritable val : values) sum += val.get(); result.se

17、t(sum); context.write(key, result);public int getPartition(K key, V value,int numReduceTasks) return (key=cstor)? 0:1;Mapper簡(jiǎn)稱:MPartition簡(jiǎn)稱:PPReduce簡(jiǎn)稱:R分布式引例分布式解決方案定義WC所需三個(gè)函數(shù)：MPR分布式引例分布式引例分布式解決方案分布式解決方案分布式存儲(chǔ)構(gòu)建 + 分布計(jì)算構(gòu)建+使用MR實(shí)現(xiàn)WC 整體流程data cstordatacstordatacstor 結(jié)果存至DFS結(jié)果存至DFS存儲(chǔ)主節(jié)點(diǎn)本地計(jì)算洗牌合并再計(jì)算存結(jié)果分布式存儲(chǔ)分

18、布式計(jì)算網(wǎng)網(wǎng) 絡(luò)絡(luò)M PRM P RM P Rfile01可靠分布式存儲(chǔ)MMM-file01可靠分布式計(jì)算第三類問題 可靠性分布式引例分布式引例分布式解決方案分布式解決方案M-file01MM03再死就把它加入黑名單冗余可靠的分布式存儲(chǔ)可靠的分布式計(jì)算思路：冗余l(xiāng)3 Sept 2008l NEUSOFT SECRETHadoop1.0-概念由來組成概念由來組成組成Hadoop1.0架構(gòu)架構(gòu)HDFS架構(gòu)架構(gòu)客戶端namenodeb-0b-1datanode1b-1b-2datanode2b-2b-0datanode3b-0b-1b-2datanoden元數(shù)據(jù)操作數(shù)據(jù)塊操作讀寫fsimageb-1

19、b-0File01分為三個(gè)塊：b-0,b-1,b-2客戶端jobtrackertasktracker1tasktracker2tasktracker3Tasktracker n提交任務(wù)分配并監(jiān)管任務(wù)JVMJVMHadoop1.0架構(gòu)架構(gòu)分布式計(jì)算架構(gòu)分布式計(jì)算架構(gòu)public class Write public static void main(String args) throws IOException Configuration conf = new Configuration(); /實(shí)例化配置文件類Path inFile = new Path(“/user/joe/myfile”)

20、; /定義欲寫文件位置FileSystem hdfs = FileSystem.get(conf); /獲取文件系統(tǒng)接口FSDataOutputStream outputStream = hdfs.create(inFile); /創(chuàng)建文件流 outputStream.writeUTF(“china cstor cstor china”); /寫入字符流outputStream.flush(); /強(qiáng)制流寫入磁盤outputStream.close(); /關(guān)閉輸出流 public class Read public static void main(String args) throws I

21、OException Configuration conf = new Configuration();Path inFile = new Path(/user/joe/myfile);FileSystem hdfs = FileSystem.get(conf);FSDataInputStream inputStream = hdfs.open(inFile);System.out.println(myfile: +inputStream.readUTF();inputStream.close(); WriteReadHadoop1.0編程編程Hdfs編程編程讀寫HdfsHdfs編程已在上頁(yè)給

22、出編程已在上頁(yè)給出Hadoop1.0編程編程 MapReduce編程編程 概述概述InputFormatMapCombinePartitionerGroupingComparatorReduceOutputFormatRecordReaderInputSplitRecordWriterOutputCommiter抽象類或接口抽象類或接口MR下默認(rèn)實(shí)現(xiàn)下默認(rèn)實(shí)現(xiàn)InputFormatTextInputFormatRecordReaderLineRecordReaderInputSplitFileSplitMapIdentityMapperCombineIdentityReducerPartiti

23、onerHashPartitionerGroupingComparator神奇的地方,參見SecondarySort.javaReduceIdentityReducerOutputFormatFileOutputFormatRecordWriterLineRecordWriterOuputCommitterFileOutputCommintterHadoop1.0編程 MapReduce編程編程過程問我們?yōu)槭裁疵看沃恍鑼憁apper和reduce因?yàn)槠渌寄J(rèn)指定了！public class WordCount public static void main(String args) thro

24、ws Exception /主方法，函數(shù)入口 Configuration conf = new Configuration(); /實(shí)例化配置文件類 Job job = new Job(conf, WordCount); /實(shí)例化Job類 job.setInputFormatClass(TextInputFormat.class); /指定使用默認(rèn)輸入格式類 TextInputFormat.setInputPaths(job, inputPaths); /設(shè)置待處理文件的位置 job.setJarByClass(WordCount.class); /設(shè)置主類名 job.setMapperCla

25、ss(TokenizerMapper.class); /指定使用上述自定義Map類 job.setMapOutputKeyClass(Text.class); /指定Map類輸出的，K類型 job.setMapOutputValueClass(IntWritable.class); /指定Map類輸出的，V類型 job.setPartitionerClass(HashPartitioner.class); /指定使用默認(rèn)的HashPartitioner類 job.setReducerClass(IntSumReducer.class); /指定使用上述自定義Reduce類 job.setNum

26、ReduceTasks(Integer.parseInt(numOfReducer); /指定Reduce個(gè)數(shù) job.setOutputKeyClass(Text.class); /指定Reduce類輸出的,K類型 job.setOutputValueClass(Text.class); /指定Reduce類輸出的,V類型 job.setOutputFormatClass(TextOutputFormat.class); /指定使用默認(rèn)輸出格式類 TextOutputFormat.setOutputPath(job, outputDir); /設(shè)置輸出結(jié)果文件位置 System.exit(j

27、ob.waitForCompletion(true) ? 0 : 1); /提交任務(wù)并監(jiān)控任務(wù)狀態(tài) Hadoop1.0編程編程之之MapReduce編編程程之之WordCount實(shí)實(shí)例例入口類入口類/*定義map類，一般繼承自Mapper類，里面實(shí)現(xiàn)讀取單詞，寫出*/ public class TokenizerMapper extends Mapper private final static IntWritable one = new IntWritable(1); private Text word = new Text(); /map方法，劃分一行文本，讀一單詞寫出一個(gè) public

28、void map(Object key, Text value, Context context)throws IOException, InterruptedException StringTokenizer itr = new StringTokenizer(value.toString(); while (itr.hasMoreTokens() word.set(itr.nextToken(); context.write(word, one);/寫出 Hadoop1.0編程之MapReduce編程之WordCount實(shí)例Mapper類/*定義類MyPartitioner，cstor到r

29、educe 0，data到reduce 1*/public class MyPartitioner extends Partitioner public int getPartition(Text key,IntWritable value,int numReduceTasks) return return (key.toString=cstor)? 0:1; Hadoop1.0編程之MapReduce編程之WordCount實(shí)例Partitioner類/*默認(rèn)Partitioner，效率高，能將數(shù)據(jù)均勻分散到各reduce，但還是可能會(huì)出現(xiàn)數(shù)據(jù)傾斜*/public class HashPar

30、titioner extends Partitioner public int getPartition(K key, V value, int numReduceTasks) return (key.hashCode() & Integer.MAX_VALUE) % numReduceTasks; /*定義reduce類，對(duì)相同的單詞，把他們中，VList值全部相加*/public static class IntSumReducer extends Reducer private IntWritable result = new IntWritable(); public void

31、 reduce(Text key, Iterable values,Context context) throws IOException, InterruptedException int sum = 0; for (IntWritable val : values) sum += val.get();/相當(dāng)于，將兩個(gè)1相加。 result.set(sum); /寫出這個(gè)單詞，和這個(gè)單詞出現(xiàn)次數(shù) context.write(key, result); Hadoop1.0編程之MapReduce編程之WordCount實(shí)例Reduce類Hadoop1.0編程編程MapReduce編程之編程之W

32、ordCount邏輯過程邏輯過程data cstordatacstordatacstor 結(jié)果存至DFS結(jié)果存至DFS存儲(chǔ)主節(jié)點(diǎn)本地計(jì)算洗牌合并再計(jì)算存結(jié)果分布式存儲(chǔ)分布式計(jì)算網(wǎng)網(wǎng) 絡(luò)絡(luò)M PRWordCount邏輯執(zhí)行過程其實(shí)就是它MapReduceProgramJobClinetJobRrackerMapTaskorReuceTaskTaskRrackerChildClient JVMClient nodejobtracker nodetasktracker node2.Get new job ID1.Runjob4.Submit job5.Initializejob3.Copy jobr

33、esources7.Heartbeat6.Retrieve input splits9.Launch10.Run8.Retrieve job resources框架執(zhí)行WordCount過程之大過程JobConfMapperReducerJobClientJob jar UrlJobInProgressTaskInProgress1TaskInProgress2TaskInputDatasJob JarOutputDatasTaskTaskInProgress1TaskRunnerMapperTaskMapperCombinerLocalFSReducerReducerTaskApplicat

34、ion1.Submit JobJobTracker2.Add New Job3.Create Task4.Launch Task5.Transfer Launch Task Action6.Add New Task7.New Runner8.Create Child11.Create Child9.InvokeMapper11.InvokeReducer10.Invoke CombinerProcessInvokeDataTaskTrackerHDFS框架執(zhí)行WC之細(xì)過程Hadoop1.0弊端弊端分布式計(jì)算HDFS工作量多而雜l3 Sept 2008l NEUSOFT SECRETHadoop

35、2.0架構(gòu)架構(gòu)概念由來組成概念由來組成組成Hadoop2.0架構(gòu)架構(gòu)Hdfs架構(gòu)架構(gòu)和原Hdfs架構(gòu)相同但可配置兩個(gè)主節(jié)點(diǎn)namenode1namenode2Zookeeper集群集群(至少至少3臺(tái)臺(tái))JournalNode集群集群(至少至少3臺(tái)臺(tái))datanode1datanode2datanoden采用這種機(jī)制可以實(shí)現(xiàn)雙采用這種機(jī)制可以實(shí)現(xiàn)雙NameNode自動(dòng)、瞬間、熱切換自動(dòng)、瞬間、熱切換原原JobTracker功能功能ResourceManagerApplicationMaster原TaskTrackerNodeManagerSlave端Master端簡(jiǎn)單的說就是簡(jiǎn)單的說就是：放權(quán)放

36、權(quán)Hadoop2.0架構(gòu)架構(gòu)Yarn架構(gòu)架構(gòu)Yarn架構(gòu)思路架構(gòu)思路：將原：將原JobTracker三三大功能拆分大功能拆分原原JobTracker功能功能ResourceManagerApplicationMasterMaster端端Hadoop2.0架構(gòu)架構(gòu)Yarn架構(gòu)架構(gòu)Yarn架構(gòu)思考問？l 既然MR邏輯由它實(shí)現(xiàn)l 任務(wù)也由它監(jiān)管那！程序可不可以不是MR類型的業(yè)務(wù)邏輯答，可以！Yarn編程，完全突破Hadoop1.0編程局限開發(fā)不同的ApplicationMaster，可實(shí)現(xiàn)不同業(yè)務(wù)邏輯如無數(shù)據(jù)只計(jì)算型，M型，MRR型Hadoop2.0架構(gòu)架構(gòu)Yarn架構(gòu)架構(gòu)Yarn架構(gòu)框架執(zhí)行Yar

37、n任務(wù)時(shí)流程Hadoop2.0架構(gòu)架構(gòu)Yarn架構(gòu)架構(gòu)強(qiáng)調(diào)l 任務(wù)邏輯完全由AppMster決定，可以不是MR邏輯l AppMster是RM啟動(dòng)的l AppMster負(fù)責(zé)任務(wù)執(zhí)行和監(jiān)管Client端向RM提交任務(wù)RM首先啟動(dòng)負(fù)責(zé)本此任務(wù)的AppMsterAppMster負(fù)責(zé)本此任務(wù)處理邏輯 它首先與RM協(xié)商申請(qǐng)資源 再NM協(xié)商啟動(dòng)Container執(zhí)行任務(wù)AppMster完成任務(wù)后向RM注銷自己Hadoop2.0編程編程概述概述Hadoop2.0編程編程Yarn編程編程Yarn編程步驟編程步驟首先編寫首先編寫Client端端其次編寫其次編寫AppMster端端Hadoop2.0編程編程Yarn

38、編程實(shí)例編程實(shí)例DistributedShell概述概述DistributedShell是Yarn自帶的一個(gè)應(yīng)用程序編程實(shí)例，相當(dāng)于Yarn編程中的“Hello World”，它的功能是并行執(zhí)行用戶提交的Shell命令或Shell腳本。定位獲取從hadoop官方網(wǎng)站下載hadoop-2.2.0-src.tar.gz（hadoop源碼包）并解壓后，依次進(jìn)入hadoop-yarn-projecthadoop-yarnhadoop-yarn-applications，下面就是Yarn自帶的兩個(gè)Yarn編程實(shí)例。DistributedShell中主要包含兩個(gè)類Client和ApplicationMas

39、ter。其中Client主要向RM提交任務(wù)。ApplicationMaster則需向RM申請(qǐng)資源，并與NM協(xié)商啟動(dòng)Container完成任務(wù)。講解執(zhí)行$hadoop jar /usr/lib/hadoop-yarn/hadoop-yarn-applications-distributedshell.jar org.apache.hadoop.yarn.applications.distributedshell.Client -jar /usr/lib/hadoop-yarn/hadoop-yarn-applications-distributedshell.jar -shell_command

40、 /bin/date -num_containers 100Client端代碼：見下頁(yè)pptApplicationMaster端代碼：見下頁(yè)ppt源碼YarnClient yarnClient = YarnClient.createYarnClient(); /新建Yarn客戶端yarnClient.start(); /啟動(dòng)Yarn客戶端YarnClientApplication app = yarnClient.createApplication(); /獲取提交程序句柄ApplicationSubmissionContext appContext = app.getApplicationS

41、ubmissionContext(); /獲取上下文句柄ApplicationId appId = appContext.getApplicationId(); /獲取RM分配的appIdappContext.setResource(capability); /設(shè)置任務(wù)其他信息舉例appContext.setQueue(amQueue); /設(shè)置作業(yè)所屬對(duì)了appContext.setPriority(priority); /設(shè)置作業(yè)優(yōu)先級(jí)/實(shí)例化ApplicationMaster對(duì)應(yīng)的ContainerContainerLaunchContext amContainer = Records.

42、newRecord(ContainerLaunchContext.class);amContainer.setCommands(commands); /參數(shù)commands為用戶預(yù)執(zhí)行的shell命令appContext.setAMContainerSpec(amContainer); /指定ApplicationMaster的ContaineryarnClient.submitApplication(appContext); /提交作業(yè)Hadoop2.0編程編程Yarn編程編程DistributedShell之之Client類分析類分析AMRMClientAsync amRMClient =

43、 AMRMClientAsync.createAMRMClientAsync(1000, allocListener); /新建RM代理amRMClient.init(conf); /設(shè)置配置文件amRMClient.start(); /獲取RM代理amRMClient.registerApplicationMaster(appMasterHostname, appMasterRpcPort,appMasterTrackingUrl); /向RM注冊(cè)NMClientAsync nmClientAsync = new NMClientAsyncImpl(containerListener); /

44、新建NM代理nmClientAsync.init(conf); /初始化NM端連接nmClientAsync.start(); /獲取NM代理for (int i = 0; i numTotalContainers; +i) /向RM申請(qǐng)資源ContainerRequest containerAsk = setupContainerAskForRM();amRMClient.addContainerRequest(containerAsk);numRequestedContainers.set(numTotalContainers); /設(shè)置任務(wù)所需全部資源量（以Containers為單位）C

45、ontainerLaunchContext ctx = Records.newRecord(ContainerLaunchContext.class); /設(shè)置Container上下文ctx.setCommands(commands); /設(shè)置NM上所啟動(dòng)的Container需完成的任務(wù)nmClientAsync.startContainerAsync(container, ctx); /要求NM啟動(dòng)containernmClientAsync.stop(); /containerListener匯報(bào)此NM完成任務(wù)后，關(guān)閉此NMamRMClient.unregisterApplicationM

46、aster(appStatus, appMessage, null); /向RM注銷amRMClient.stop(); /任務(wù)完成，停止RM端代理Hadoop2.0編程Yarn編程(DistributedShell之ApplicationMaster類分析)Hadoop2.0編程編程Yarn編程編程MapReduce編程編程Yarn下的MapReduce和MRv1里的MapReduce相同，是Google-MapReduce思想的實(shí)現(xiàn)，不同的是，Yarn下編程不只有MR這一種模式，如果我們將MR看成一種編程模板，那么MRv1下，則只有這一種模板，可是Yarn下可以有各種各樣的編程模板，如無數(shù)據(jù)只耗CPU計(jì)算形式，如MR形式，盡管編程形式多了，但MR依舊是分布式編程首選模板。定位同DistributedShell相同，編寫MapReduce程序基本步驟依舊是： 編寫Client：默認(rèn)實(shí)現(xiàn)類MRClientService 編寫ApplicationMaster：默認(rèn)實(shí)現(xiàn)類MRAppMasterHadoop開發(fā)人員為MapReduce編程模型開發(fā)了Client和ApplicationMaster的默認(rèn)實(shí)現(xiàn)了MRClientServe和MRAppMaster，MRv1版本的MapReduce代碼在Yarn里是兼容的，用戶只需重新編譯一下