Hadoop是一個(gè)十分流行的分布式存儲(chǔ)和計(jì)算框架，也是業(yè)內(nèi)大數(shù)據(jù)處理和分析最通用的框架之一。
Hadoop2.0 由HDFS（Hadoop Distributed File System）搅裙、MapReduce和Yarn三部分組成摘刑。
Hadoop的設(shè)計(jì)原型來(lái)源于google的三篇論文，即GFS、MapReduce和BigTable,同時(shí)作為L(zhǎng)ucene的子項(xiàng)目Nutch的一部分在2005年引入Apache,Hadoop的得名是Hadoop之父Doug Cutting兒子的玩具蝗茁，值得一提的是他的妻子叫Lucene醋虏。
Hadoop生態(tài)
"永遠(yuǎn)不把雞蛋放在一個(gè)籃子里"——HDFS
在分布式的文件系統(tǒng)之前，往往使用大型機(jī)和存儲(chǔ)服務(wù)器去做存儲(chǔ)哮翘，無(wú)論大型機(jī)和存儲(chǔ)服務(wù)器都是十分昂貴的颈嚼，同時(shí)也是有瓶頸的，橫向擴(kuò)展能力很差饭寺。而分布式文件系統(tǒng)的橫向擴(kuò)展能力以及容錯(cuò)性十分好阻课，也越來(lái)越受到人們的青睞。HDFS的定位是用比較廉價(jià)的機(jī)器艰匙，做高可用的海量數(shù)據(jù)的存儲(chǔ)限煞。主要采用多副本的分塊存儲(chǔ)機(jī)制，在部分機(jī)器宕機(jī)或數(shù)據(jù)損壞的情況下员凝，依然能提供可靠服務(wù)署驻。
- 集群拓?fù)?/li>
NameNode:文件元數(shù)據(jù)存儲(chǔ)節(jié)點(diǎn)，Hadoop1中存在單點(diǎn)問(wèn)題健霹，Hadoop2中通過(guò)備用節(jié)點(diǎn)實(shí)現(xiàn)高可用旺上，同時(shí)有元數(shù)據(jù)存儲(chǔ)瓶頸，不適于存儲(chǔ)小文件糖埋。
DataNode:數(shù)據(jù)節(jié)點(diǎn)宣吱，單文件被分成多塊，每一塊多副本跨機(jī)架阶捆、機(jī)房存儲(chǔ)凌节，保證數(shù)據(jù)的高可用。
Block:文件的存儲(chǔ)單位洒试，單個(gè)文件可被分成多塊倍奢，默認(rèn)為128M。同時(shí)也是MapReduce默認(rèn)的輸入塊大小垒棋。
文件讀取流程：
open DistributedFileSystem 去NameNode獲取文件的塊列表卒煞，NameNode根據(jù)Client距離各節(jié)點(diǎn)的網(wǎng)絡(luò)距離給出Block列表。
根據(jù)Block列表去一次讀取文件叼架，讀取后在Client進(jìn)行文件匯總畔裕。
- 寫(xiě)入流程：
實(shí)例化DistributeFileSystem，在NameNode進(jìn)行寫(xiě)文件的申請(qǐng)乖订，申請(qǐng)成功后創(chuàng)建元數(shù)據(jù)扮饶，并返回?cái)?shù)據(jù)存放的位置信息。
過(guò)位置信息乍构，對(duì)DataNode進(jìn)行流式寫(xiě)入甜无，將數(shù)據(jù)分成多個(gè)數(shù)據(jù)包，作為一個(gè)數(shù)據(jù)隊(duì)列。寫(xiě)入時(shí)每次取一個(gè)數(shù)據(jù)包,寫(xiě)入全部副本,且三個(gè)節(jié)點(diǎn)均寫(xiě)入成功岂丘，則返回ACK信號(hào)表名當(dāng)前buff寫(xiě)入成功陵究，Client內(nèi)部維護(hù)著數(shù)據(jù)包的Ack隊(duì)列，收到Ack之后會(huì)移除這個(gè)數(shù)據(jù)包奥帘。
最后想NameNode發(fā)送Complete信號(hào)铜邮，確認(rèn)文件寫(xiě)入全部成功。
如果中途節(jié)點(diǎn)寫(xiě)失斦！：寫(xiě)入部分?jǐn)?shù)據(jù)的節(jié)點(diǎn)將在管線(xiàn)中移除松蒜，同時(shí)后續(xù)恢復(fù)正常后會(huì)刪除這部分?jǐn)?shù)據(jù)。隨后寫(xiě)入后續(xù)兩個(gè)節(jié)點(diǎn)钥庇，原則上寫(xiě)入一個(gè)塊的時(shí)候就可以寫(xiě)成功牍鞠，因?yàn)閚amenode發(fā)現(xiàn)數(shù)據(jù)不一致會(huì)做復(fù)制操作。
"分而治之"——MapReduce###
</br>
分布式計(jì)算出現(xiàn)之前评姨，數(shù)據(jù)的計(jì)算往往依靠性能比較好的單機(jī)計(jì)算难述。但是單機(jī)受限于本身的計(jì)算資源，往往計(jì)算速度都不如人意吐句。
一天小明接到產(chǎn)品的一個(gè)需求：
產(chǎn)品：小明啊胁后，這里有一天的日志信息，大概5個(gè)G嗦枢，我要統(tǒng)計(jì)一下一共有多少攀芯。
小明：OK啊，就5個(gè)G文虏，一個(gè)shell搞定侣诺，看我 cat * | wc -l，我簡(jiǎn)直就是個(gè)天才氧秘。
產(chǎn)品：對(duì)不起啊小明年鸳，需求變了，一天的看不出來(lái)效果丸相，我需要統(tǒng)計(jì)1個(gè)月的數(shù)據(jù)搔确，大概有150G。
小明：有點(diǎn)大啊灭忠，不怕膳算，我線(xiàn)上服務(wù)器內(nèi)存120G，40核弛作，看我用多線(xiàn)程搞定涕蜂，過(guò)了2個(gè)小時(shí)，終于搞 定了還有點(diǎn)費(fèi)勁映琳。
產(chǎn)品：我保證這是我最后一次變更需求宇葱，我想要最近一年的數(shù)據(jù)1800G左右瘦真。
小明：數(shù)據(jù)上T了，搞不定了啊黍瞧。
上面的例子告訴我們，在大數(shù)量的場(chǎng)景下原杂，高性能的單機(jī)有時(shí)也是解決不了問(wèn)題印颤。所以我們就需要MapReduce幫助我們。
</br>
MapReduce是一種采用分治和規(guī)約的一種并行的批處理框架穿肄，先將數(shù)據(jù)做分割計(jì)算年局，最后匯總結(jié)果∠滩看上去和多線(xiàn)程的處理機(jī)制一樣矢否，但是Hadoop將它封裝在了框架中，編程十分簡(jiǎn)單脑溢，吞吐量十分高僵朗，目前支持Java、C++屑彻、Python等多種API編程验庙。
１.MapReduce運(yùn)行模型總體概覽:
InputSplit： InputSplit是單個(gè)map任務(wù)的輸入文件片，默認(rèn)文件的一個(gè)block社牲。
Map函數(shù)：數(shù)據(jù)處理邏輯的主體粪薛，用開(kāi)發(fā)者開(kāi)發(fā)。
Partition：map的結(jié)果發(fā)送到相應(yīng)的reduce搏恤。
Combain：reduce之前進(jìn)行一次預(yù)合并违寿，減小網(wǎng)絡(luò)IO。當(dāng)然熟空，部分場(chǎng)景不適合藤巢。
Shuffle：map輸出數(shù)據(jù)按照Partition分發(fā)到各個(gè)reduce。
*reduce：將不同map匯總來(lái)的數(shù)據(jù)做reduce邏輯痛阻。
２.多reduce：
３.經(jīng)典wordcount：
４.Map類(lèi)的實(shí)現(xiàn):
必須繼承org.apache.hadoop.mapreduce.Mapper 類(lèi)
map()函數(shù),對(duì)于每一個(gè)輸入K/V都會(huì)調(diào)用一次map函數(shù)菌瘪，邏輯實(shí)現(xiàn)(必須)。
setup()函數(shù),在task開(kāi)始前調(diào)用一次阱当，做maptask的一些初始化工作俏扩，如連接數(shù)據(jù)庫(kù)、加載配置(可選)弊添。
cleanup()函數(shù),在task結(jié)束前調(diào)用一次录淡，做maptask的收尾清理工作,如批處理的收尾，關(guān)閉連接等（可選）
Context上下文環(huán)境對(duì)象油坝，包含task相關(guān)的配置嫉戚、屬性和狀態(tài)等刨裆。
5.Reduce類(lèi)的實(shí)現(xiàn):
必須繼承org.apache.hadoop.mapreduce.Reducer類(lèi)。
reduce(key, Iterable<>values,Context context)對(duì)于每一個(gè)key值調(diào)用一次reduce函數(shù)彬檀。
setup():在task開(kāi)始前調(diào)用一次帆啃，做reducetask的一些初始化工作。
cleanup()：在task結(jié)束時(shí)調(diào)用一次窍帝，做reducetask的收尾清理工作努潘。
6.作業(yè)整體配置:
參數(shù)解析:String[]otherArgs= new GenericOptionsParser(conf, args).getRemainingArgs();
創(chuàng)建job: Jobjob= Job.getInstance(conf, "word count");
設(shè)置map類(lèi)，reduce類(lèi)坤学。
設(shè)置map和reduce輸出的KV類(lèi)型疯坤，二者輸出類(lèi)型一致的話(huà)則可以只設(shè)置Reduce的輸出類(lèi)型。
設(shè)置reduce的個(gè)數(shù) :默認(rèn)為1深浮，綜合考慮压怠，建議單個(gè)reduce處理數(shù)據(jù)量<10G。不想啟用reduce設(shè)置為0即可飞苇。
設(shè)置InputFormat
設(shè)置OutputFormat
設(shè)置輸入菌瘫，輸出路徑。
job.waitForCompletion(true) (同步提交)和job.submit()(異步提交)
wordcount：
public class WordCountTask { private static final Logger logger = Logger.getLogger(WordCountTask.class); public static class WordCountMap extends Mapper<Object, Text, Text, IntWritable>{ private static final IntWritable one = new IntWritable(1); private Text word = new Text(); @Override protected void cleanup(Context context) throws IOException, InterruptedException { logger.info("mapTaskEnd....."); } protected void map(Object key, Text value, Context context) throws IOException, InterruptedException { StringTokenizer itr = new StringTokenizer(value.toString()); while (itr.hasMoreTokens()) { this.word.set(itr.nextToken()); context.write(this.word, one); } } protected void setup(Context context) throws IOException, InterruptedException { logger.info("mapTaskStart....."); } } public static class WordCountReduce extends Reducer<Text, IntWritable, Text, IntWritable>{ private IntWritable result = new IntWritable(); public void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException { int sum = 0; for (IntWritable val : values) { sum += val.get(); } this.result.set(sum); context.write(key, this.result); } } public static void main(String[] args) throws Exception { Configuration conf = new Configuration(); String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs(); if (otherArgs.length < 2) { System.err.println("Usage: wordcount <in> [<in>...] <out>"); System.exit(2); } Job job = Job.getInstance(conf, "word count"); job.setJarByClass(WordCountTask.class); job.setMapperClass(WordCountMap.class); job.setReducerClass(WordCountReduce.class); job.setOutputKeyClass(Text.class); job.setOutputValueClass(IntWritable.class); FileSystem fs = FileSystem.get(conf); for (int i = 0; i < otherArgs.length - 1; i++) { FileInputFormat.addInputPath(job, new Path(otherArgs[i])); } if(fs.exists(new Path(otherArgs[otherArgs.length - 1]))){ fs.delete(new Path(otherArgs[otherArgs.length - 1])); } FileOutputFormat.setOutputPath(job, new Path(otherArgs[(otherArgs.length - 1)])); job.setNumReduceTasks(1); System.exit(job.waitForCompletion(true) ? 0 : 1); } }
6.提交：
?hadoop jar hadoop-examples.jar demo.wordcount(主類(lèi)名) Dmapreduce.job.queuename=XX(系統(tǒng)參數(shù)) input output
?缺點(diǎn)：無(wú)定時(shí)調(diào)度
- 常用的InputFormat:
TextInputFormat key:行偏移 value:文本內(nèi)容玄柠，split計(jì)算：splitSize=max("mapred.min.split.size",min("mapred.max.split.size",blockSize)) mapred.min.split.size 在大量文本輸入的情況下突梦，需要控制map的數(shù)量，可以調(diào)此選項(xiàng)羽利。
CombineTextInputFormat（集群默認(rèn)）宫患，多個(gè)小文件分片送到一個(gè)map中處理，主要解決多個(gè)小文件消耗map資源的問(wèn)題这弧。
sequenceFileInputFormat娃闲，采用自己的序列化方式，通常文件名為key匾浪，value為文件內(nèi)容皇帮，可在存儲(chǔ)上解決小文件對(duì)namenode的影響。
