hadoop實(shí)戰(zhàn)培訓(xùn)傳智播客

1、高級(jí)軟件人才實(shí)作培訓(xùn)專家!Hahoop 培訓(xùn)1海量數(shù)據(jù)處理平臺(tái)架構(gòu)一般網(wǎng)站把用戶的訪問行為記錄以apach 日志的形式記錄下來了，這些日志中包含了下面一些 關(guān)鍵字段： client_ip ,user_id,access_time,url,referer,status,page_size,agent因?yàn)樾枰y(tǒng)一對(duì)數(shù)據(jù)進(jìn)行離線計(jì)算，所以常常把它們?nèi)恳频酵粋€(gè)地方。簡(jiǎn)單算了一下： （1） 網(wǎng)站請(qǐng)求數(shù)：1kw/天 （2） 每天日志大?。?50Byte/行* 1kw = 4.2G， （3） 日志存儲(chǔ)周期：2 年 一天產(chǎn)生4.5G 的日志，2 年需要4.2G * 2 * 365 = 3.0T解決方案：

2、為了方便系統(tǒng)命令查看日志，不壓縮，總共需要3.0T 的空間，剛好有一些2U 的服務(wù)器，每臺(tái)共1T 的磁盤空間。為了避免系統(tǒng)盤壞掉影響服務(wù)器使用，對(duì)系統(tǒng)盤做了raid1。為了避免其他存放數(shù)據(jù)的盤壞掉導(dǎo)致數(shù)據(jù)無法恢復(fù)，對(duì)剩下的盤做了raid5。所有的數(shù)據(jù)都匯聚到這幾臺(tái)LogBackup 服務(wù)器上來了。2 有了LogBackup 服務(wù)器，離線統(tǒng)計(jì)就可以全部在這些服務(wù)器上進(jìn)行了。在這套架構(gòu)上，用wc、grep、sort、uniq、awk、sed 等系統(tǒng)命令，完成了很多的統(tǒng)計(jì)需求，比如統(tǒng)計(jì)訪問頻率較高的client_ip，某個(gè)新上線的的頁面的referer 主要是哪些網(wǎng)站。 當(dāng)業(yè)務(wù)的迅猛發(fā)展，網(wǎng)站流量爆

3、發(fā)增長(zhǎng)，產(chǎn)品經(jīng)理如果想從中獲取更多的用戶特征和用戶信息， 就需要我們這些數(shù)據(jù)分析人員從不同的日志中找到令他們滿意的答案。如果 （1） 日志總行數(shù)：10 億/天 （2） 每天日志大?。?50Byte/行* 10 億= 420G， （3） 日志種類：5 種3Hadoop能解決哪些問題海量數(shù)據(jù)需要及時(shí)分析和處理。海量數(shù)據(jù)需要深入分析和挖掘。數(shù)據(jù)需要長(zhǎng)期保存問題：磁盤IO成為一種瓶頸,而非CPU資源。網(wǎng)絡(luò)帶寬是一種稀缺資源硬件故障成為影響穩(wěn)定的一大因素4Hadoop在國(guó)內(nèi)的情景奇虎360：Hadoop存儲(chǔ)軟件管家中軟件，使用CDN技術(shù)將用戶請(qǐng)求引到最近的Hadoop集群并進(jìn)行下載京東、百度：存儲(chǔ)、分析

4、日志、數(shù)據(jù)挖掘和機(jī)器學(xué)習(xí)（主要是推薦系統(tǒng)）廣告類公司：存儲(chǔ)日志，通過協(xié)調(diào)過濾算法為客戶推薦廣告Yahoo：垃圾郵件過濾華為：云計(jì)算平臺(tái)Facebook：日志存儲(chǔ)，實(shí)時(shí)分析某公安部項(xiàng)目：網(wǎng)民QQ聊天記錄與關(guān)聯(lián)人調(diào)查系統(tǒng)，使用Hbase實(shí)現(xiàn)某學(xué)校：學(xué)生上網(wǎng)與社會(huì)行為分析，使用hadoop淘寶、阿里：國(guó)內(nèi)使用Hadoop最深入的公司，整個(gè)Taobao和阿里都是數(shù)據(jù)驅(qū)動(dòng)的5Hadoop開發(fā)人員市場(chǎng)需求和待遇6Hadoop在國(guó)內(nèi)的人才儲(chǔ)備7Hadoop介紹1）作者：Doug Cutting2)用Java編寫的開源系統(tǒng)，能夠安排在大規(guī)模的計(jì)算平臺(tái)上，從而長(zhǎng)進(jìn)計(jì)算效率。3）Nutch搜索引擎里面的一個(gè)模塊。

5、4）受Google三篇論文的啟發(fā) MapReduce GFS Bigtable5）google hadoop mapreduce mapreduce GFS HDFS Bigtable Hbase8Hadoop生態(tài)系統(tǒng)介紹Hbase 1）Nosql數(shù)據(jù)庫，Key-Value存儲(chǔ) 2）最大化利用內(nèi)存HDFS 1） hadoop distribute 分布式文件系統(tǒng) 2）最大化利用磁盤MapReduce 1）編程模型，主要用來做數(shù)據(jù)的分析 2）最大化利用CPU9 HDFS篇10HDFS設(shè)計(jì)原則文件以塊(block)方式存儲(chǔ)每個(gè)塊帶下遠(yuǎn)比多數(shù)文件系統(tǒng)來的大(預(yù)設(shè)64M)通過副本機(jī)制提高可靠度和讀取吞

6、吐量每個(gè)區(qū)塊至少分到三臺(tái)DataNode上單一 master (NameNode)來協(xié)調(diào)存儲(chǔ)元數(shù)據(jù)(metadata)客戶端對(duì)文件沒有緩存機(jī)制 (No data caching)11HDFS系統(tǒng)結(jié)構(gòu)12NameNode(NN)NameNode主要功能提供名稱查詢服務(wù)，它是一個(gè)jetty服務(wù)器NameNode保存metadate信息包括文件owership和permissions文件包含哪些塊Block保存在哪個(gè)DataNode（由DataNode啟動(dòng)時(shí)上報(bào)）NameNode的metadate信息在啟動(dòng)后會(huì)加載到內(nèi)存metadata存儲(chǔ)到磁盤文件名為”fsimage”Block的位置信息不會(huì)保存

7、到fsimage13NameNode塊存儲(chǔ)結(jié)構(gòu)metadate物理存儲(chǔ)結(jié)構(gòu)14DataNode（DN）保存Block啟動(dòng)DN線程的時(shí)候會(huì)向NN匯報(bào)block信息通過向NN發(fā)送心跳保持與其聯(lián)系（3秒一次），如果NN 10分鐘沒有收到DN的心跳，則認(rèn)為其已經(jīng)lost，并copy其上的block到其它DN 15Block的副本放置策略第一個(gè)副本：放置在上傳文件的DN；如果是集群外提交，則隨機(jī)挑選一臺(tái)磁盤不太滿，CPU不太忙的節(jié)點(diǎn)第二個(gè)副本：放置在于第一個(gè)副本不同的機(jī)架的節(jié)點(diǎn)上第三個(gè)副本：與第二個(gè)副本相同集群的節(jié)點(diǎn)更多副本：隨機(jī)節(jié)點(diǎn)16再說Block設(shè)置一個(gè)Block 64MB，如果上傳文件小于該值，

8、仍然會(huì)占用一個(gè)Block的命名空間（NameNode metadata），但是物理存儲(chǔ)上不會(huì)占用64MB的空間Block大小和副本數(shù)由Client端上傳文件到HDFS時(shí)設(shè)置，其中副本數(shù)可以變更，Block是不可以再上傳后變更的17數(shù)據(jù)損壞(corruption)處理當(dāng)DN讀取block的時(shí)候，它會(huì)計(jì)算checksum如果計(jì)算后的checksum，與block創(chuàng)建時(shí)值不一樣，說明該block已經(jīng)損壞。client讀取其它DN上的block；NN標(biāo)記該塊已經(jīng)損壞，然后復(fù)制block達(dá)到預(yù)期設(shè)置的文件備份數(shù)DN在其文件創(chuàng)建后三周驗(yàn)證其checksum18HDFS文件權(quán)限與Linux文件權(quán)限類似r:

9、read; w:write; x:execute，權(quán)限x對(duì)于文件忽略，對(duì)于文件夾表示是否允許訪問其內(nèi)容如果Linux系統(tǒng)用戶zhangsan使用hadoop命令創(chuàng)建一個(gè)文件，那么這個(gè)文件在HDFS中owner就是zhangsanHDFS的權(quán)限目的：阻止好人錯(cuò)錯(cuò)事，而不是阻止壞人做壞事。HDFS相信，你告訴我你是誰，我就認(rèn)為你是誰19HDFS文件讀取20HDFS文件寫入21HDFS文件存儲(chǔ) 兩個(gè)文件，一個(gè)文件156M，一個(gè)文件128在HDFS里面怎么存儲(chǔ)？ -Block為64MB -rapliction默認(rèn)拷貝3份22HDFS文件存儲(chǔ)結(jié)構(gòu)23HDFS開發(fā)常用命令創(chuàng)建一個(gè)文件夾？上傳一個(gè)文件？刪除

10、一個(gè)文件和文件夾？查看一個(gè)文件夾里面有哪些文件？查看某個(gè)文件的內(nèi)容？24Hadoop管理員常用命令hadoop job list #列出正在運(yùn)行的Jobhadoop job kill #kill jobhadoop fsck / #檢查HDFS塊狀態(tài)，是否損壞hadoop fsck / -delete #檢查HDFS塊狀態(tài)，刪除損壞塊hadoop dfsadmin report #檢查HDFS狀態(tài)，包括DN信息hadoop dfsadmin safemode enter | leavehadoop distcp hdfs:/a:8020/xxx hdfs:/b:8020/ #并行copy./b

11、in/start-balancer.sh #平衡集群文件25HDFS API詳解static get(Configuration conf)operator() /step1 得到Configuration對(duì)象 /step2 得到對(duì)象 /step3 進(jìn)行文件操作 26用Java對(duì)HDFS編程文件操作1.1上傳本地文件到hadoop fs1.2 在hadoop fs中新建文件，并寫入1.3 刪除hadoop fs上的文件1.4讀取文件1.5 文件修改時(shí)間2. 目錄操作2.1 在hadoop fs上創(chuàng)建目錄2.2 刪除目錄2.3 讀取某個(gè)目錄下的所有文件2.4遍歷hdfshdfs信息查找某個(gè)文件在

12、HDFS集群的位置獲取HDFS集群上所有節(jié)點(diǎn)名稱信息27 mapreduce篇28開發(fā)hadoop依賴的jar和自帶的example到 下載hadoop-0.20.2.tar.gz，目前最新版為1.0.3。下載完后解壓文件,y有hadoop-0.20.2-core.jar, hadoop-0.20.2-examples.jar,hadoop-0.20.2-core.jar :hadoop的核心類庫Hadoop所依賴的jar： hadoop所依賴的jar在lib目錄下面。Hadoop 自帶的一些案例分析： hadoop-0.20.2-examples.jar是hadoop-0.20.2自帶的一些

13、案例。介紹如下： 1) aggregatewordcount計(jì)算輸入文件中文字個(gè)數(shù)的基于聚合的MapReduce程序。 2) aggregatewordhist生成輸入文件中文字個(gè)數(shù)的統(tǒng)計(jì)圖的基于聚合的MapReduce程序。 3) grep計(jì)算輸入文件中匹配正則表達(dá)式的文字個(gè)數(shù)的MapReduce程序。4) join合并排序的平均分割的數(shù)據(jù)集的作業(yè)。5)pentomino解決五格拼版問題的分塊分層的MapReduce程序。29Hadoop自帶的examples.jar介紹6)pi使用蒙地卡羅法計(jì)算PI的MapReduce程序。7)Randomtextwriter在一個(gè)節(jié)點(diǎn)上寫10G隨機(jī)文本的

14、MapReduce程序。8)randomwriter在每個(gè)節(jié)點(diǎn)上寫10G隨機(jī)數(shù)據(jù)的MapReduce程序。9) sleep在每個(gè)Map和Reduce作業(yè)中休憩的程序。10)sort排序隨機(jī)寫入器生成的數(shù)據(jù)的MapReduce程序。11)sudoku一個(gè)九宮格游戲的解決方案。12)wordcount在輸入文件中統(tǒng)計(jì)文字個(gè)數(shù)的統(tǒng)計(jì)器。30實(shí)例寫MapReduce程序的步驟：1.把問題轉(zhuǎn)化為MapReduce模型2.設(shè)置運(yùn)行的參數(shù)3.寫map類 4.寫reduce類例子：統(tǒng)計(jì)單詞個(gè)數(shù) My name is liu gangWhat is your nameMy 1name 2is 2What 1yo

15、ur 1liu 1gang 131MapReduce模型1.Map端一行行讀文件，程序轉(zhuǎn)化為中間Key/Value. My name is liu gang -My 1,name 1, is 1, liu 1, gang 1 What is your name -What 1, is 1, your 1 name 12.Reduce端相同的key肯定會(huì)在一起。經(jīng)過Reduce方法處理后，形成最終的key/Value. name 1,name 1-name 2;32運(yùn)行步驟1）打成jar包。2）創(chuàng)建一個(gè)word.txt文件3）把word.txt文件傳到HDFS上面 hadoop fs copyF

16、romLocal 4)執(zhí)行hadoop jar 5）查看執(zhí)行結(jié)果 hadoop fs text /path 33MapReduce執(zhí)行流程34MapReduce基本流程35JobTracker(JT)和TaskTracker(TT)簡(jiǎn)介36再論JobTracker(JT)和TaskTracker(TT)JobTracker：協(xié)作作業(yè)的運(yùn)行taskTracker：運(yùn)行作業(yè)劃分后的任務(wù)37Mapreduce原理1）一個(gè)文件2）存儲(chǔ)文件3) 統(tǒng)計(jì)文件里面”Refund”個(gè)數(shù)38JobTracker失敗1）JobTracker失敗在所有的失敗中是最嚴(yán)重的一種。2）hadoop沒有處理jobtracke

17、r失敗的機(jī)制。-它是一個(gè)單點(diǎn)故障。3）在未來的新版本中可能可以運(yùn)行多個(gè)JobTracker。4）可以使用ZooKeeper來協(xié)作JobTracker。39TaskTracker失敗1）一個(gè)TaskTracker由于崩潰或運(yùn)行過于緩慢而失敗，它會(huì)向JobTracker發(fā)送“心跳”。2）如果有未完成的作業(yè)，JobTracker會(huì)重新把這些任務(wù)分配到其他的TaskTracker上面運(yùn)行。3）即使TaskTracker沒有失敗也可以被JobTracker列入黑名單。40Hadoop Job Scheduler1）Hadoop默認(rèn)的調(diào)度器是基于隊(duì)列的FIFO調(diào)度器。 所有用戶的作業(yè)都被提交到一個(gè)隊(duì)列中，

18、然后由JobTracker先按照作業(yè)的優(yōu)先級(jí)高低，再按照作業(yè)提交時(shí)間的先后順序選擇將被執(zhí)行的作業(yè)。優(yōu)點(diǎn): 調(diào)度算法簡(jiǎn)單明了，JobTracker工作負(fù)擔(dān)輕。缺點(diǎn): 忽略了不同作業(yè)的需求差異。2）還用兩個(gè)多用戶調(diào)度器 -Fair Scheduler(公平調(diào)度器) -Capacity Scheduler(計(jì)算能力調(diào)度) 41Fair Scheduler(公平調(diào)度器)多個(gè)Pool，Job需要被提交到某個(gè)Pool中每個(gè)pool可以設(shè)置最小 task slot，稱為miniShareFS會(huì)保證Pool的公平Pool內(nèi)部支持Priority設(shè)置支持資源搶占42 mapreduce的類型與格式Hadoop

19、0.20.x之前的APIHadoop的MapReduce中，map和reduce函數(shù)遵循如下常規(guī)格式： map: (K1, V1) list(K2, V2) reduce: (K2, list(V2) list(K3, V3)Mapper的接口： public interface Mapper extends JobConfigurable, Closeable void map(K1 key, V1 value, OutputCollector output, Reporter reporter) throws IOException; Reduce的接口： public interface

20、 Reducer extends JobConfigurable, Closeable void reduce(K2 key, Iterator values, OutputCollector output, Reporter reporter) throws IOException; /outputCollector 是為了輸出key/value對(duì)，/Reporter 是用來更新計(jì)數(shù)和狀態(tài)信息。43Hadoop 0.20.x之后的APIHadoop的MapReduce中，map和reduce函數(shù)遵循如下常規(guī)格式： map: (K1, V1) list(K2, V2) reduce: (K2,

21、 list(V2) list(K3, V3)Mapper的接口：protected void map(KEY key, VALUE value, Context context) throws IOException, InterruptedException Reduce的接口：protected void reduce(KEY key, Iterable values, Context context ) throws IOException, InterruptedException /Context是上下文對(duì)象，這里Context等同于OutputCollector和Reporter兩

22、個(gè)函數(shù)的功能。44mapreduce的數(shù)據(jù)類型與java類型對(duì)應(yīng)關(guān)系Java的基本類型Writable實(shí)現(xiàn)booleanBooleanWritablebyteByteWritableintIntWritablefloatFloatWritablelongLongWritabledoubleDoubleWritableStringText45Writable接口1.對(duì)Java中的int型進(jìn)行封裝那么就是hadoop中的IntWritable類在寫程序時(shí)可以把IntWritable可以看著是int 類型，它實(shí)現(xiàn) 了WritableComparable接口。 WritableComparable又是

23、Writable、parable接口的子接口。2.Writable類對(duì)所有的Java基本類型進(jìn)行封裝：如：boolean - BooleanWritable;Byte - ByteWritable3. 我們可以自定義Writable接口，來編寫更復(fù)雜的結(jié)構(gòu)的類。核心：hadoop有自己一套的I/O機(jī)制。I/O類都必須實(shí)現(xiàn)Writable接口。46實(shí)現(xiàn)自定義的mapreduce類型public class LogKey implements WritableComparable private String customerId; private String dateTime; private

24、 String domain;public void readFields(DataInput in) throws IOException customerId = in.readUTF(); dateTime = in.readUTF(); domain = in.readUTF();public void write(DataOutput out) throws IOException out.writeUTF(customerId); out.writeUTF(dateTime); out.writeUTF(domain);/當(dāng)我們?cè)趯?shí)際開發(fā)的時(shí)候會(huì)遇到多個(gè)key的時(shí)候，mapredu

25、ce自帶的類型是不能滿足我們的需求，這樣我們就要自己來定制化Key和Value。47最小的MapReduce驅(qū)動(dòng)public class MinimalMapReduceWithDefaults extends Configured implements Tool public int run(String args) throws IOException JobConf conf = JobBuilder.parseInputAndOutput(this, getConf(), args);if (conf = null) return -1;conf.setInputFormat(Text

26、InputFormat.class);conf.setNumMapTasks(1);conf.setMapperClass(IdentityMapper.class);conf.setMapRunnerClass(MapRunner.class);conf.setMapOutputKeyClass(LongWritable.class);conf.setMapOutputValueClass(Text.class);conf.setPartitionerClass(HashPartitioner.class);conf.setNumReduceTasks(1);conf.setReducerC

27、lass(IdentityReducer.class);conf.setOutputKeyClass(LongWritable.class);conf.setOutputValueClass(Text.class);conf.setOutputFormat(TextOutputFormat.class);JobClient.runJob(conf);return 0;public static void main(String args) throws Exception int exitCode = ToolRunner.run(new MinimalMapReduceWithDefault

28、s(), args);System.exit(exitCode);48mapreduce驅(qū)動(dòng)默認(rèn)的設(shè)置InputFormat(輸入)TextInputFOrmatMapperClass(map類)IdentityMapperMapRunnerClass(map啟動(dòng)類)MapRunnerMapOutputKeyClassLongWritableMapOutputValueClassTextPartitionerClassHashPartitionerReduceClassIdentityReduceOutputKeyClassLongWritableOutputValueClassTextOut

29、putFormatClassTextOutputFormat49Combiners和Partitioner編程Combiners的作用： 每一個(gè)map可能會(huì)產(chǎn)生大量的輸出，combiner的作用就是在map端對(duì)輸出先做一次合并，以減少傳輸?shù)絩educer的數(shù)據(jù)量， 1）combiner最基本是實(shí)現(xiàn)本地key的聚合，對(duì)map輸出的key排序，value進(jìn)行迭代。如下所示：map: (K1, V1) list(K2, V2)combine: (K2, list(V2) list(K2, V2)reduce: (K2, list(V2) list(K3, V3) 2）combiner還具有類似本地的

30、reduce功能.例如hadoop自帶的wordcount的例子和找出value的最大值的程序，combiner和reduce完全一致。如下所示：map: (K1, V1) list(K2, V2)combine: (K2, list(V2) list(K3, V3)reduce: (K3, list(V3) list(K4, V4) 3）如果不用combiner，那么，所有的結(jié)果都是reduce完成，效率會(huì)相對(duì)低下。使用combiner，先完成的map會(huì)在本地聚合，提升速度。 4）對(duì)于hadoop自帶的wordcount的例子，value就是一個(gè)疊加的數(shù)字，所以map一結(jié)束就可以進(jìn)行redu

31、ce的value疊加，而不必要等到所有的map結(jié)束再去進(jìn)行reduce的value疊加。注意：combiner使用的合適，可以在滿足業(yè)務(wù)的情況下提升job的速度，如果不合適，則將導(dǎo)致輸出的結(jié)果不正確。Combiner的輸出是Reducer的輸入，Combiner絕不能改變最終的計(jì)算結(jié)果。所以從我的想法來看，Combiner只應(yīng)該用于那種Reduce的輸入key/value與輸出key/value類型完全一致，且不影響最終結(jié)果的場(chǎng)景。比如累加，最大值等。50 Combiners分析假設(shè)有兩個(gè)map。第一個(gè)map的輸出為：（1950,0） (1950,20) (1950,10)第二個(gè)map輸出為：

32、（1950,25） (1950,15) (1950,30)Reduce函數(shù)被調(diào)用是，輸入如下：（1950，0,20,10,25,15,30）因?yàn)?0是最大的值，所以輸出如下：（1950,30）如果我們使用 combiner：那么reduce調(diào)用的時(shí)候傳入的數(shù)據(jù)如下：（1950，20,30）-（1950,30）用表達(dá)式表示為：Max(0,20,10,25,15,30)=max(max(0,20,10),max(25,15,30)=max(20,30)=30 51使用 Combiners要小心剛才我們是計(jì)算最大值可以使用Combiners能提高效率。如果我們要是求平均值呢？Avg（0,20,10,25,15,30） = 15如果使用Combiner會(huì)得到什么樣的結(jié)果呢？第一個(gè)map輸出為： avg(0,20,10) = 10第二個(gè)map輸出為：Avg（25,15,30） = 23輸入到reduce出來的結(jié)果為：Avg(10,23) = 17.517.5和15？所以 ：使用combiner一定要注意。52 Partitioner分析Partitioner 在mapreduce的位置：53mapreduce提供的PatitionerPartition主要作用就是將