1. MAPREDUCE使用

mapreduce是hadoop中的分布式運(yùn)算編程框架，只要按照其編程規(guī)范，只需要編寫少量的業(yè)務(wù)邏輯代碼即可實(shí)現(xiàn)一個(gè)強(qiáng)大的海量數(shù)據(jù)并發(fā)處理程序

?

2. Demo開發(fā)——wordcount

2.1需求

從大量（比如T級(jí)別）文本文件中，統(tǒng)計(jì)出每一個(gè)單詞出現(xiàn)的總次數(shù)。

?

2.2mapreduce 實(shí)現(xiàn)思路

Map階段：

a)?從HDFS的源數(shù)據(jù)文件中逐行讀取數(shù)據(jù)

b)?將每一行數(shù)據(jù)切分出單詞

c)?為每一個(gè)單詞構(gòu)造一個(gè)鍵值對(duì)(單詞，1)

d)?將鍵值對(duì)發(fā)送給reduce

?

Reduce階段：

a)?接收map階段輸出的單詞鍵值對(duì)

b)?將相同單詞的鍵值對(duì)匯聚成一組

c)?對(duì)每一組，遍歷組中的所有“值”，累加求和，即得到每一個(gè)單詞的總次數(shù)

d)?將(單詞，總次數(shù))輸出到HDFS的文件中

?

2.3具體編碼實(shí)現(xiàn)

(1)定義一個(gè)mapper類

?

//首先要定義四個(gè)泛型的類型 //keyin: LongWritable valuein: Text //keyout: Text valueout:IntWritablepublic class WordCountMapper extends Mapper<LongWritable, Text, Text, IntWritable>{//map方法的生命周期： 框架每傳一行數(shù)據(jù)就被調(diào)用一次//key : 這一行的起始點(diǎn)在文件中的偏移量//value: 這一行的內(nèi)容 @Overrideprotected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {//拿到一行數(shù)據(jù)轉(zhuǎn)換為stringString line = value.toString(); //將這一行切分出各個(gè)單詞String[] words = line.split(" ");//遍歷數(shù)組，輸出<單詞，1>for(String word:words){context.write(new Text(word), new IntWritable(1));}} }

?

(2)定義一個(gè)reducer類

?

//生命周期：框架每傳遞進(jìn)來一個(gè)kv 組，reduce方法被調(diào)用一次 @Overrideprotected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {//定義一個(gè)計(jì)數(shù)器int count = 0;//遍歷這一組kv的所有v，累加到count中for(IntWritable value:values){count += value.get();}context.write(key, new IntWritable(count));} }

?

(3)定義一個(gè)主類，用來描述job并提交job

?

public class WordCountRunner {//把業(yè)務(wù)邏輯相關(guān)的信息（哪個(gè)是mapper，哪個(gè)是reducer，要處理的數(shù)據(jù)在哪里，輸出的結(jié)果放哪里。。。。。。）描述成一個(gè)job對(duì)象//把這個(gè)描述好的job提交給集群去運(yùn)行public static void main(String[] args) throws Exception {Configuration conf = new Configuration();Job wcjob = Job.getInstance(conf);//指定我這個(gè)job所在的jar包 // wcjob.setJar("/home/hadoop/wordcount.jar");wcjob.setJarByClass(WordCountRunner.class);wcjob.setMapperClass(WordCountMapper.class);wcjob.setReducerClass(WordCountReducer.class);//設(shè)置我們的業(yè)務(wù)邏輯Mapper類的輸出key和value的數(shù)據(jù)類型wcjob.setMapOutputKeyClass(Text.class);wcjob.setMapOutputValueClass(IntWritable.class);//設(shè)置我們的業(yè)務(wù)邏輯Reducer類的輸出key和value的數(shù)據(jù)類型wcjob.setOutputKeyClass(Text.class);wcjob.setOutputValueClass(IntWritable.class); //指定要處理的數(shù)據(jù)所在的位置FileInputFormat.setInputPaths(wcjob, "hdfs://hdp-server01:9000/wordcount/data/big.txt");//指定處理完成之后的結(jié)果所保存的位置FileOutputFormat.setOutputPath(wcjob, new Path("hdfs://hdp-server01:9000/wordcount/output/"));//向yarn集群提交這個(gè)jobboolean res = wcjob.waitForCompletion(true);System.exit(res?0:1);}

?

3.程序打包運(yùn)行

?1. 將程序打包

?2. 準(zhǔn)備輸入數(shù)據(jù)

vi??/home/hadoop/test.txt

Hello tom Hello jim Hello ketty Hello world Ketty tom

在hdfs上創(chuàng)建輸入數(shù)據(jù)文件夾：

hadoop???fs ?mkdir ?-p ?/wordcount/input

將words.txt上傳到hdfs上

hadoop??fs ?–put ?/home/hadoop/words.txt ?/wordcount/input

3. 將程序jar包上傳到集群的任意一臺(tái)服務(wù)器上

4. 使用命令啟動(dòng)執(zhí)行wordcount程序jar包

$ hadoop jar wordcount.jar cn.itcast.bigdata.mrsimple.WordCountDriver /wordcount/input /wordcount/out

5.?查看執(zhí)行結(jié)果

$ hadoop?fs –cat /wordcount/out/part-r-00000

?

作者：杰瑞教育
出處：http://www.cnblogs.com/jerehedu/?
版權(quán)聲明：本文版權(quán)歸煙臺(tái)杰瑞教育科技有限公司和博客園共有，歡迎轉(zhuǎn)載，但未經(jīng)作者同意必須保留此段聲明，且在文章頁面明顯位置給出原文連接，否則保留追究法律責(zé)任的權(quán)利。
技術(shù)咨詢：

?

總結(jié)

以上是生活随笔為你收集整理的Java+大数据开发——Hadoop集群环境搭建(二)的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。