Hadoop及Mapreduce入門檸檬學(xué)院大數(shù)據(jù)培訓(xùn)在線教育OutlinesHadoopOverviewHDFSMap-reduceProgrammingParadigmHadoopMap-reduceJobSchedulerResourcesHadoop,Why?數(shù)據(jù)太多了，需要能存儲、快速分析Pb級數(shù)據(jù)集的平臺單機的存儲、IO、CPU有限，需要可擴展的集群單點故障問題 –單點故障是正?，F(xiàn)象，但需要處理這種異常 –節(jié)點有增有減需要好用的、可靠的基礎(chǔ)平臺來解決，最好是開源的HadoopHistoryDec2004–GoogleGFSpaperpublishedJuly2005–NutchusesMapReduceFeb2006–BecomesLucenesubprojectApr2007–Yahoo!on1000-nodeclusterApr2008–FastestQuickSorton1TBJan2008–AnApacheTopLevelProjectJul2008–A4000nodetestclusterHadoop-relatedprojectsNutch：網(wǎng)頁搜索軟件，不只是爬蟲Avro：數(shù)據(jù)序列化系統(tǒng)Chukwa：用于管理大規(guī)模分布式集群的數(shù)據(jù)收集系統(tǒng)ZooKeeper：用于分布式應(yīng)用的高性能協(xié)同服務(wù)Hbase：類似于BigTable的，key-value數(shù)據(jù)庫系統(tǒng)PIG：并行計算的一種高級語言Hive：數(shù)據(jù)倉庫Mahout：分布式機器學(xué)習(xí)和數(shù)據(jù)挖掘的LibHama：基于BSP的超大規(guī)?？茖W(xué)計算框架WhoUsesHadoopAmazonAdobeAdknowledge：

behavioraltargeting,clickstreamanalyticsAlibabaBaidu:搜索日志分析；每周處理3000TB數(shù)據(jù)BixoLabs：webminingDatagraph：處理RDF數(shù)據(jù)，存儲、索引EBay

：532nodes，搜索優(yōu)化和研究ETHZurichSystemsGroup：教學(xué)《MassivelyParallelDataAnalysiswithMapReduce》Facebook

：1100nodes,12PB;300nodes,3PBFOX：3個Cluster用于日志分析、數(shù)據(jù)挖掘、機器學(xué)習(xí)Freestylers：構(gòu)建基于圖片的推薦系統(tǒng)GoogleGruter.Corp：索引、Link分析、數(shù)據(jù)挖掘Hulu：HbasehostingIBMKrugle

：源代碼搜索Last.fm：圖表計算、A/B測試，userprofile分析，cookie級的報表處理LinebergerComprehensiveCancerCenter：癌癥相關(guān)的研究，使用SeqWareLinkedIn：這人你可能認識TheNewYorkTimesPARC：分析Wikipedia里的沖突Powerset/MicrosoftPressflip：個性化搜索，訓(xùn)練SVM模型Yahoo?。?000nodes(2*4cpuboxesw4*1TBdisk&16GBRAM)Moreon/hadoop/PoweredBy

GoalsofHDFS大數(shù)據(jù)集存儲 –10Knodes,100millionfiles,10PB

應(yīng)付硬件故障 –用文件多副本應(yīng)付故障問題 –故障自動檢測和恢復(fù)更適合批量處理 –搬移計算比搬移數(shù)據(jù)更廉價 –數(shù)據(jù)一次寫入，多次讀取 –更注重數(shù)據(jù)讀取的高吞吐量，而不是低延時適應(yīng)復(fù)雜的硬件及軟件平臺TheFileSystem一個集群只有一個Namespace跟Unix的文件系統(tǒng)Namespace很相似，不過不支持Hardlink、Softlink文件分塊存儲一般一塊大小為64M，可配置每塊會被復(fù)制在多個DataNode上支持回收站當(dāng)一個文件被刪除時會先放入用戶下的回收站回收站會被定期清除恢復(fù)的方式是將文件從回收站移出NameNodeMetadataMeta-data存在內(nèi)存中 –整個Meta-data放入主內(nèi)存 –Nodemandpagingofmeta-dataMeta-data記錄了 –文件列表信息 –每個文件的塊列表 –每個塊對應(yīng)的DataNode –文件屬性，如創(chuàng)建時間、創(chuàng)建者、幾份副本等TransactionLog(EditLog) –記錄了文件系統(tǒng)的每個變化，如創(chuàng)建文件、刪除文件、修改文件的副本數(shù)等 –EditLog會被合并為FsImage并存入磁盤Meta-data磁盤故障 –NameNode可以維護多份數(shù)據(jù)DataNodeABlockServer –將數(shù)據(jù)存儲本機文件系統(tǒng)(e.g.ext3) –存儲數(shù)據(jù)塊的Meta-data(e.g.CRC,ID)匯報 –啟動時向NameNode注冊本地存儲的文件塊 –定期向NameNode報告本機存活（心跳）數(shù)據(jù)輸送 –接收來自客戶端的寫數(shù)據(jù) –向客戶端發(fā)送數(shù)據(jù) –將數(shù)據(jù)傳輸?shù)街付ǖ腄ataNodesBlockReplicaPlacement機架感知NameNode能感知機架，選擇較優(yōu)的方式假設(shè)有3份或以上，目前的策略是一份放在本地節(jié)點上第二份放在另外一個機架的節(jié)點上第三份放在跟第二份同機架的不同節(jié)點上其他的隨機放置客戶端從最近的塊讀取DataCorrectness用CRC32來做數(shù)據(jù)校驗 –UseCRC32文件寫入 –由客戶端負責(zé)計算CRC –checksum存放在DataNode文件讀取 –客戶端從DataNode讀取數(shù)據(jù)和checksum –由客戶端校驗，如果不通過，則客戶端嘗試從其他的副本讀取數(shù)據(jù)FSShellFShadoopfs-mkdir/foodirhadoopfs-rmr/foodirhadoopfs-cat/foodir/myfile.txthadoopfs-tail/foodir/myfile.txtChmod，chown，put，mv，cp，du，dus更多命令請運行hadoopfs–help獲取DFSAdminSafemode，upgradeProgress，refreshNodes，…fsck文件系統(tǒng)檢查Balancer集群均衡WebUIMap-reduceProgrammingParadigmMap-reduceProgrammingParadigmMap-reduce是一種適合分布式計算的編程范式input|map|shuffle|reduce|output最簡單的實現(xiàn)方式cat*|grep‘java’|sort|uniq-c|cat>file實現(xiàn)這種編程范式的有GoogleHadoopOracleTeradata……HadoopMap/Reduce(input)<k1,v1>->

map

-><k2,v2>->

combine*

-><k2,v2>->reduce

-><k3,v3>(output)combine過程可能沒有，也可能有多次Mapperpublic

classMapper<KEYIN,VALUEIN,KEYOUT,VALUEOUT>{

/***Calledonceatthestartofthetask.*/

protected

voidsetup(Contextcontext)throwsIOException,InterruptedException{}

protected

voidmap(KEYINkey,VALUEINvalue,Contextcontext)throwsIOException,InterruptedException{context.write((KEYOUT)key,(VALUEOUT)value);}

/***Calledonceattheendofthetask.*/

protected

voidcleanup(Contextcontext)throwsIOException,InterruptedException{}}Reducerpublic

classReducer<KEYIN,VALUEIN,KEYOUT,VALUEOUT>{

/***Calledonceatthestartofthetask.*/

protected

voidsetup(Contextcontext)throwsIOException,InterruptedException{}

protected

voidreduce(KEYINkey,Iterable<VALUEIN>values,Contextcontext)throwsIOException,InterruptedException{

for(VALUEINvalue:values){context.write((KEYOUT)key,(VALUEOUT)value);

}

}

/***Calledonceattheendofthetask.*/

protected

voidcleanup(Contextcontext)throwsIOException,InterruptedException{}}JobSetup

public

static

voidmain(String[]args)throwsException{String[]otherArgs=newGenericOptionsParser(conf,args).getRemainingArgs();

if(otherArgs.length!=2){System.err.println("Usage:wordcount<in><out>");System.exit(2);

}Jobjob=newJob(newConfiguration(),"wordcount");job.setJarByClass(WordCount.class);

//設(shè)置輸入job.setInputFormatClass(TextInputFormat.class);FileInputFormat.addInputPath(job,newPath(otherArgs[0]));

//Mapjob.setMapperClass(TokenizerMapper.class);

//Combinejob.setCombinerClass(IntSumReducer.class);

//Reducejob.setReducerClass(IntSumReducer.class);

job.setOutputKeyClass(Text.class);job.setOutputValueClass(IntWritable.class);

//設(shè)置輸出FileOutputFormat.setOutputPath(job,newPath(otherArgs[1]));

//提交Job并執(zhí)行System.exit(job.waitForCompletion(true)?0:1);

}InjectPointsInputjob.setInputFormatClass()SplitRecordReaderMapjob.setMapperClass()Combine*job.setCombinerClass()Shufflingjob.setPartitionerClass()Sortjob.setSortComparatorClass()Groupingjob.setGroupingComparatorClass()Reducejob.setReducerClass()Outputjob.setOutputFormatClass()JobTracker&TaskTrackerJob&Task一個Job會被分成多個Task執(zhí)行一個Task對應(yīng)一個Map或者ReduceJobTracker運行在Master上，管理和跟蹤每個Job收集Task的信息狀態(tài)，并匯總重新調(diào)度失敗的任務(wù)TaskTracker向JobTracker匯報狀態(tài)（心跳）運行在每個計算節(jié)點上，管理和跟蹤每個Task收集task的信息，并提供給JobTrackerFIFO先到先得，排隊執(zhí)行FairSchedule