2.hadoop從入門到精通第3天筆記第三天

1、北京傳智播客教育 Hadoop深入淺出北京傳智播客教育 北京傳智播客教育 課程安排MapReduce原理*MapReduce執(zhí)行過(guò)程*數(shù)據(jù)類型與格式*Writable接口與序列化機(jī)制*-加深拓展-MapReduce的執(zhí)行過(guò)程源碼分析問(wèn)題：怎樣解決海量數(shù)據(jù)的計(jì)算？求和：1+5+7 +3+4+9 +3+5+6=?北京傳智播客教育 1+5+73+4+93+5+6Map:13 + 16 + 14Reduce:MapReduce概述MapReduce是一種分布式計(jì)算模型，由Google提出，主要用于搜索領(lǐng)域，解決海量數(shù)據(jù)的計(jì)算問(wèn)題.MR由兩個(gè)階段組成：Map和Reduce，用戶只需要實(shí)現(xiàn)map()和re

2、duce()兩個(gè)函數(shù)，即可實(shí)現(xiàn)分布式計(jì)算，非常簡(jiǎn)單。這兩個(gè)函數(shù)的形參是key、value對(duì)，表示函數(shù)的輸入信息。北京傳智播客教育 Mapreduce原理北京傳智播客教育 執(zhí)行步驟： 1. map任務(wù)處理1.1 讀取輸入文件內(nèi)容，解析成key、value對(duì)。對(duì)輸入文件的每一行，解析成key、value對(duì)。每一個(gè)鍵值對(duì)調(diào)用一次map函數(shù)。1.2 寫自己的邏輯，對(duì)輸入的key、value處理，轉(zhuǎn)換成新的key、value輸出。2.reduce任務(wù)處理2.1 在reduce之前，有一個(gè)shuffle的過(guò)程對(duì)多個(gè)map任務(wù)的輸出進(jìn)行合并、排序。2.2寫reduce函數(shù)自己的邏輯，對(duì)輸入的key、valu

3、e處理，轉(zhuǎn)換成新的key、value輸出。2.3 把reduce的輸出保存到文件中。例子：實(shí)現(xiàn)WordCountApp北京傳智播客教育 map、reduce鍵值對(duì)格式函數(shù)輸入鍵值對(duì)輸出鍵值對(duì)map()reduce()北京傳智播客教育 WordCountApp的驅(qū)動(dòng)代碼public static void main(String args) throws Exception Configuration conf = new Configuration(); /加載配置文件 Job job = new Job(conf); /創(chuàng)建一個(gè)job，供JobTracker使用 job.setJarByCl

4、ass(WordCountApp.class); job.setMapperClass(WordCountMapper.class); job.setReducerClass(WordCountReducer.class); FileInputFormat.setInputPaths(job, new Path(hdfs:/192.168.1.10:9000/input); FileOutputFormat.setOutputPath(job, new Path(hdfs:/192.168.1.10:9000/output); job.setOutputKeyClass(Text.class)

5、; job.setOutputValueClass(IntWritable.class); pletion(true);北京傳智播客教育 MR流程代碼編寫作業(yè)配置提交作業(yè)初始化作業(yè)分配任務(wù)執(zhí)行任務(wù)更新任務(wù)和狀態(tài)完成作業(yè) runjar clientRMHDFSClusterwc.jardatadatadatadataNMJVM(child)北京傳智播客教育 MR過(guò)程各個(gè)角色的作用jobClient：提交作業(yè)JobTracker：初始化作業(yè)，分配作業(yè)，TaskTracker與其進(jìn)行通信，協(xié)調(diào)監(jiān)控整個(gè)作業(yè)TaskTracker：定期與JobTracker通信，執(zhí)行Map和Reduce任務(wù)HDFS：保

6、存作業(yè)的數(shù)據(jù)、配置、jar包、結(jié)果北京傳智播客教育 作業(yè)提交提交作業(yè)之前，需要對(duì)作業(yè)進(jìn)行配置編寫自己的MR程序配置作業(yè)，包括輸入輸出路徑等等提交作業(yè)配置完成后，通過(guò)JobClient提交具體功能與JobTracker通信得到一個(gè)jar的存儲(chǔ)路徑和JobId輸入輸出路徑檢查將jobj ar拷貝到的HDFS計(jì)算輸入分片，將分片信息寫入到j(luò)ob.split中寫job.xml真正提交作業(yè)北京傳智播客教育 作業(yè)初始化客戶端提交作業(yè)后，JobTracker會(huì)將作業(yè)加入到隊(duì)列，然后進(jìn)行調(diào)度，默認(rèn)是FIFO方式具體功能作業(yè)初始化主要是指JobInProgress中完成的讀取分片信息創(chuàng)建task包括Map和Re

7、duce任創(chuàng)建task包括Map和Reduce任務(wù)創(chuàng)建TaskInProgress執(zhí)行task，包括map任務(wù)和reduce任務(wù)北京傳智播客教育 任務(wù)分配TaskTracker與JobTracker之間的通信和任務(wù)分配是通過(guò)心跳機(jī)制實(shí)現(xiàn)的TaskTracker會(huì)主動(dòng)定期向JobTracker發(fā)送心態(tài)信息，詢問(wèn)是否有任務(wù)要做，如果有，就會(huì)申請(qǐng)到任務(wù)。北京傳智播客教育 任務(wù)執(zhí)行如果TaskTracker拿到任務(wù)，會(huì)將所有的信息拷貝到本地，包括代碼、配置、分片信息等TaskTracker中的localizeJob()方法會(huì)被調(diào)用進(jìn)行本地化，拷貝job.jar，jobconf，job.xml到本地Ta

8、skTracker調(diào)用launchTaskForJob()方法加載啟動(dòng)任務(wù)MapTaskRunner和ReduceTaskRunner分別啟動(dòng)java child進(jìn)程來(lái)執(zhí)行相應(yīng)的任務(wù)北京傳智播客教育 狀態(tài)更新Task會(huì)定期向TaskTraker匯報(bào)執(zhí)行情況TaskTracker會(huì)定期收集所在集群上的所有Task的信息，并向JobTracker匯報(bào)JobTracker會(huì)根據(jù)所有TaskTracker匯報(bào)上來(lái)的信息進(jìn)行匯總北京傳智播客教育 作業(yè)完成JobTracker是在接收到最后一個(gè)任務(wù)完成后，才將任務(wù)標(biāo)記為成功將數(shù)結(jié)果據(jù)寫入到HDFS中北京傳智播客教育 錯(cuò)誤處理JobTracker失敗存在單點(diǎn)

9、故障，hadoop2.0解決了這個(gè)問(wèn)題TraskTracker失敗TraskTracker崩潰了會(huì)停止向JobTracker發(fā)送心跳信息。JobTracker會(huì)將TraskTracker從等待的任務(wù)池中移除，并將該任務(wù)轉(zhuǎn)移到其他的地方執(zhí)行JobTracker將TaskTracker加入到黑名單中Task失敗任務(wù)失敗，會(huì)向TraskTracker拋出異常任務(wù)掛起北京傳智播客教育 JobTracker負(fù)責(zé)接收用戶提交的作業(yè)，負(fù)責(zé)啟動(dòng)、跟蹤任務(wù)執(zhí)行。JobSubmissionProtocol是JobClient與JobTracker通信的接口。InterTrackerProtocol是TaskTra

10、cker與JobTracker通信的接口。北京傳智播客教育 TaskTracker負(fù)責(zé)執(zhí)行任務(wù)。北京傳智播客教育 JobClient是用戶作業(yè)與JobTracker交互的主要接口。負(fù)責(zé)提交作業(yè)的，負(fù)責(zé)啟動(dòng)、跟蹤任務(wù)執(zhí)行、訪問(wèn)任務(wù)狀態(tài)和日志等。北京傳智播客教育 北京傳智播客教育 序列化概念序列化（Serialization）是指把結(jié)構(gòu)化對(duì)象轉(zhuǎn)化為字節(jié)流。反序列化（Deserialization）是序列化的逆過(guò)程。即把字節(jié)流轉(zhuǎn)回結(jié)構(gòu)化對(duì)象。Java序列化（java.io.Serializable）北京傳智播客教育 Hadoop序列化的特點(diǎn)序列化格式特點(diǎn)：緊湊：高效使用存儲(chǔ)空間?？焖伲鹤x寫數(shù)據(jù)的額外

11、開銷小可擴(kuò)展：可透明地讀取老格式的數(shù)據(jù)互操作：支持多語(yǔ)言的交互Hadoop的序列化格式：Writable北京傳智播客教育 Hadoop序列化的作用序列化在分布式環(huán)境的兩大作用：進(jìn)程間通信，永久存儲(chǔ)。Hadoop節(jié)點(diǎn)間通信。消息序列化為二進(jìn)制流節(jié)點(diǎn)1二進(jìn)制流反序列化為消息節(jié)點(diǎn)2二進(jìn)制流消息北京傳智播客教育 Writable接口Writable接口, 是根據(jù) DataInput 和 DataOutput 實(shí)現(xiàn)的簡(jiǎn)單、有效的序列化對(duì)象.MR的任意Key和Value必須實(shí)現(xiàn)Writable接口.MR的任意key必須實(shí)現(xiàn) parable接口北京傳智播客教育 常用的Writable實(shí)現(xiàn)類Text一般認(rèn)為它

12、等價(jià)于java.lang.String的Writable。針對(duì)UTF-8序列。例:Text test = new Text(test);IntWritable one = new IntWritable(1);北京傳智播客教育 自定義Writable類實(shí)現(xiàn) parable.Java值對(duì)象的比較：一般需要重寫toString(),hashCode(),equals()方法Writablewrite 是把每個(gè)對(duì)象序列化到輸出流readFields是把輸入流字節(jié)反序列化北京傳智播客教育 自定義WritableKpi電信例子把上面例子里的Mapper的value改寫為自定義Writable類型。修改原

13、MapReduce程序，并成功執(zhí)行。結(jié)果跟原來(lái)一致。北京傳智播客教育 MapReduce輸入的處理類FileInputFormat: FileInputFormat是所有以文件作為數(shù)據(jù)源的InputFormat實(shí)現(xiàn)的基類，F(xiàn)ileInputFormat保存作為job輸入的所有文件，并實(shí)現(xiàn)了對(duì)輸入文件計(jì)算splits的方法。至于獲得記錄的方法是有不同的子類TextInputFormat進(jìn)行實(shí)現(xiàn)的。北京傳智播客教育 InputFormatInputFormat 負(fù)責(zé)處理MR的輸入部分.有三個(gè)作用:驗(yàn)證作業(yè)的輸入是否規(guī)范.把輸入文件切分成InputSplit.提供RecordReader 的實(shí)現(xiàn)類，

14、把InputSplit讀到Mapper中進(jìn)行處理.北京傳智播客教育 InputSplit 在執(zhí)行mapreduce之前，原始數(shù)據(jù)被分割成若干split，每個(gè)split作為一個(gè)map任務(wù)的輸入，在map執(zhí)行過(guò)程中split會(huì)被分解成一個(gè)個(gè)記錄（key-value對(duì)），map會(huì)依次處理每一個(gè)記錄。 FileInputFormat只劃分比HDFS block大的文件，所以FileInputFormat劃分的結(jié)果是這個(gè)文件或者是這個(gè)文件中的一部分. 如果一個(gè)文件的大小比block小，將不會(huì)被劃分，這也是Hadoop處理大文件的效率要比處理很多小文件的效率高的原因。 當(dāng)Hadoop處理很多小文件（文件大

15、小小于hdfs block大?。┑臅r(shí)候，由于FileInputFormat不會(huì)對(duì)小文件進(jìn)行劃分，所以每一個(gè)小文件都會(huì)被當(dāng)做一個(gè)split并分配一個(gè)map任務(wù)，導(dǎo)致效率底下。 例如：一個(gè)1G的文件，會(huì)被劃分成16個(gè)64MB的split，并分配16個(gè)map任務(wù)處理，而10000個(gè)100kb的文件會(huì)被10000個(gè)map任務(wù)處理。 北京傳智播客教育 TextInputFormat TextInputformat是默認(rèn)的處理類，處理普通文本文件。 文件中每一行作為一個(gè)記錄，他將每一行在文件中的起始偏移量作為key，每一行的內(nèi)容作為value。 默認(rèn)以n或回車鍵作為一行記錄。 TextInputForma

16、t繼承了FileInputFormat。北京傳智播客教育 InputFormat類的層次結(jié)構(gòu)北京傳智播客教育 其他輸入類 CombineFileInputFormat 相對(duì)于大量的小文件來(lái)說(shuō)，hadoop更合適處理少量的大文件。 CombineFileInputFormat可以緩解這個(gè)問(wèn)題，它是針對(duì)小文件而設(shè)計(jì)的。 KeyValueTextInputFormat 當(dāng)輸入數(shù)據(jù)的每一行是兩列，并用tab分離的形式的時(shí)候，KeyValueTextInputformat處理這種格式的文件非常適合。 NLineInputformat NLineInputformat可以控制在每個(gè)split中數(shù)據(jù)的行數(shù)。

17、 SequenceFileInputformat 當(dāng)輸入文件格式是sequencefile的時(shí)候，要使用SequenceFileInputformat作為輸入。北京傳智播客教育 自定義輸入格式1）繼承FileInputFormat基類。2）重寫里面的getSplits(JobContext context)方法。3）重寫createRecordReader(InputSplit split, TaskAttemptContext context)方法。(講解源代碼)北京傳智播客教育 Hadoop的輸出 TextOutputformat 默認(rèn)的輸出格式，key和value中間值用tab隔開的。 SequenceFileOutputformat 將key和value以sequencefile格式輸出。 SequenceFileAsOutputFormat 將key和value以原始