Hadoop怎么實(shí)現(xiàn)HelloWorld

這篇文章給大家分享的是有關(guān)Hadoop怎么實(shí)現(xiàn)HelloWorld的內(nèi)容。小編覺得挺實(shí)用的，因此分享給大家做個(gè)參考，一起跟隨小編過來看看吧。

作為一家“創(chuàng)意+整合+營(yíng)銷”的成都網(wǎng)站建設(shè)機(jī)構(gòu)，我們?cè)跇I(yè)內(nèi)良好的客戶口碑。創(chuàng)新互聯(lián)提供從前期的網(wǎng)站品牌分析策劃、網(wǎng)站設(shè)計(jì)、做網(wǎng)站、成都網(wǎng)站制作、創(chuàng)意表現(xiàn)、網(wǎng)頁制作、系統(tǒng)開發(fā)以及后續(xù)網(wǎng)站營(yíng)銷運(yùn)營(yíng)等一系列服務(wù)，幫助企業(yè)打造創(chuàng)新的互聯(lián)網(wǎng)品牌經(jīng)營(yíng)模式與有效的網(wǎng)絡(luò)營(yíng)銷方法,創(chuàng)造更大的價(jià)值。

先從源碼看起，再一步步剖析

package org.apache.hadoop.examples;

import java.io.IOException;
import java.util.StringTokenizer;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;

public class WordCount {

  public static class TokenizerMapper 
       extends Mapper{

    private final static IntWritable one = new IntWritable(1);
    private Text word = new Text();

    public void map(Object key, Text value, Context context
                    ) throws IOException, InterruptedException {
      StringTokenizer itr = new StringTokenizer(value.toString());
      while (itr.hasMoreTokens()) {
        word.set(itr.nextToken());
        context.write(word, one);
      }
    }
  }

  public static class IntSumReducer 
       extends Reducer {
    private IntWritable result = new IntWritable();

    public void reduce(Text key, Iterable values, 
                       Context context
                       ) throws IOException, InterruptedException {
      int sum = 0;
      for (IntWritable val : values) {
        sum += val.get();
      }
      result.set(sum);
      context.write(key, result);
    }
  }

  public static void main(String[] args) throws Exception {
    Configuration conf = new Configuration();
    conf.set("mapred.job.tracker", "172.16.10.15:9001");//自己額外加的代碼
    String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
    if (otherArgs.length != 2) {
      System.err.println("Usage: wordcount  ");
      System.exit(2);
    }
    Job job = new Job(conf, "word count");
    job.setJarByClass(WordCount.class);
    job.setMapperClass(TokenizerMapper.class);
    job.setCombinerClass(IntSumReducer.class);
    job.setReducerClass(IntSumReducer.class);
    job.setOutputKeyClass(Text.class);
    job.setOutputValueClass(IntWritable.class);
    FileInputFormat.addInputPath(job, new Path(otherArgs[0]));
    FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));
    System.exit(job.waitForCompletion(true) ? 0 : 1);
  }
}

大家可以看到整個(gè)源代碼分為三個(gè)部分：

1. Map

public static class TokenizerMapper extends Mapper{

    private final static IntWritable one = new IntWritable(1);
    private Text word = new Text();

    public void map(Object key, Text value, Context context) 
throws IOException, InterruptedException {
      StringTokenizer itr = new StringTokenizer(value.toString());
      while (itr.hasMoreTokens()) {
        word.set(itr.nextToken());
        context.write(word, one);
      }
    }
  }

a) 定義一個(gè)自己的Map過程，TokenizerMapper 這個(gè)類名自己設(shè)定，這個(gè)類需要繼承org.apache.hadoop.mapreduce包中的Mapper類，四個(gè)參數(shù)分別表示輸入鍵key的參數(shù)類型，輸入值value的參數(shù)類型，輸出鍵key的參數(shù)類型，輸出值value的參數(shù)類型。 值得注意的是Hadoop本身提供了一套可優(yōu)化的網(wǎng)絡(luò)序列化傳輸?shù)幕绢愋停皇怯胘ava內(nèi)嵌的類型。這些類型都是在org.apache.hadoop.io包中。其中LongWritable類型相當(dāng)于Long類型，Text類型相當(dāng)于String類型，IntWritable相當(dāng)于Integer類型。
b) map方法中參數(shù)value是指文本文件中的一行，參數(shù)key是為該行首字母相對(duì)于文本文件首地址的偏移量
c) StringTokenizer類是一個(gè)用來分隔String的應(yīng)用類，類似于split。

//它的構(gòu)造函數(shù)有三種：
public StringTokenizer(String str)
public StringTokenizer(String str,String delim)
public StringTokenizer(String str,String delim,boolean returnDelims)
//其中第一個(gè)參數(shù)為要分隔的String,第二個(gè)參數(shù)為分隔字符集合，第三個(gè)參數(shù)為分隔符是否作為標(biāo)記返回，如果不指定分隔符，默認(rèn)是'\t\n\r\f'
//它的方法主要有三種：
public boolean hasMoreTokens()//返回是否還有分隔符
public String nextToken()//返回從當(dāng)前位置到下一個(gè)分隔符的字符串
public int countTokens()//返回nextToken方法被調(diào)用的次數(shù)

d) 經(jīng)過StringTolenizer 處理之后會(huì)得到一個(gè)個(gè) < word,1 > 這樣的鍵值對(duì)，放在context里，Context用于輸出內(nèi)容的寫入，讀起來有點(diǎn)兒繞口，自己理解一下。

2. Reduce

public static class IntSumReducer extends Reducer {
	private IntWritable result = new IntWritable();

	public void reduce(Text key, Iterable values, Context context) throws IOException, InterruptedException {
		int sum = 0;
		for (IntWritable val : values) {
			sum += val.get();
		}
		result.set(sum);
		context.write(key, result);
	}
}

a) 同mapper 過程一樣，Reduce過程需要繼承org.apache.hadoop.mapreduce包中Reducer類，并重寫其reduce方法。 
b) reduce方法中輸入?yún)?shù)key 指單個(gè)單詞，values 指對(duì)應(yīng)單詞的計(jì)數(shù)值的列表 
c) reduce 方法的目的就是對(duì)列表的值進(jìn)行加和處理 
d) 輸出的是< key,value>,key 指單個(gè)單詞，value 指對(duì)應(yīng)單詞的計(jì)數(shù)值的列表的值的總和。

3. Main

public static void main(String[] args) throws Exception {
	Configuration conf = new Configuration();
	conf.set("mapred.job.tracker", "172.16.10.15:9001");//自己額外加的代碼
	String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
	if (otherArgs.length != 2) {
		System.err.println("Usage: wordcount  ");
		System.exit(2);
	}
	Job job = new Job(conf, "word count");
	job.setJarByClass(WordCount.class);
	job.setMapperClass(TokenizerMapper.class);
	job.setCombinerClass(IntSumReducer.class);
	job.setReducerClass(IntSumReducer.class);
	job.setOutputKeyClass(Text.class);
	job.setOutputValueClass(IntWritable.class);
	FileInputFormat.addInputPath(job, new Path(otherArgs[0]));
	FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));
	System.exit(job.waitForCompletion(true) ? 0 : 1);
}

a) Configuration conf = new Configuration(); 默認(rèn)情況下，Configuration開始實(shí)例化的時(shí)候，會(huì)從Hadoop的配置文件里讀取參數(shù)。 
b) conf.set(“mapred.job.tracker”, “172.16.10.15:9001”);設(shè)置這句代碼是由于我們要把使用eclipse提交作業(yè)到Hadoop集群，所以手動(dòng)添加Job運(yùn)行地址。若是直接在Hadoop 集群進(jìn)行運(yùn)行，不用加這句代碼。 而且你可以看到只要前三句使用了這個(gè)代碼，所以這三句以后的代碼才是所有Hadoop例子中都會(huì)包含的。
c) 接下來這一句也是讀取參數(shù)，這里是從命令行參數(shù)里讀取參數(shù)。 
d) Job job = new Job(conf, “word count”); 在MapReduce處理過程中，由Job對(duì)象負(fù)責(zé)管理和運(yùn)行一個(gè)計(jì)算任務(wù)，然后通過Job的若干方法來對(duì)任務(wù)的參數(shù)進(jìn)行設(shè)置。”word count”是Job的名字，(當(dāng)然了，根據(jù)所有java語言規(guī)范規(guī)定的那樣，你也可以用

Job job = new Job();
job.setJobName("Name");

的形式做聲明)。 
e) job.setJarByClass(WordCount.class);是根據(jù)WordCount類的位置設(shè)置Jar文件 。

為什么要這么做？因?yàn)槲覀冊(cè)贖adoop集群上運(yùn)行這個(gè)作業(yè)時(shí)候，要把代碼打包成一個(gè)JAR文件，用以在集群上發(fā)布這個(gè)文件。Hadoop利用這個(gè)傳遞進(jìn)去的類來查找包含它的JAR文件。
f) job.setMapperClass(TokenizerMapper.class);設(shè)置Mapper 
g) job.setCombinerClass(IntSumReducer.class);設(shè)置Combiner,這里先使用Reduce類來進(jìn)行Mapper 的中間結(jié)果的合并，能夠減輕網(wǎng)絡(luò)傳輸?shù)膲毫Α?nbsp;
h) job.setReducerClass(IntSumReducer.class);設(shè)置Reduce 
i) job.setOutputKeyClass(Text.class);和 job.setOutputValueClass(IntWritable.class);分別是設(shè)置輸出鍵的類型和設(shè)置輸出值的類型 
j) FileInputFormat.addInputPath(job, new Path(otherArgs[0]));設(shè)置輸入文件，它是otherArgs第一個(gè)參數(shù) 
k) FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));設(shè)置輸出文件，將輸出結(jié)果寫入這個(gè)文件里，它是otherArgs第二個(gè)參數(shù) 。

注意：在運(yùn)行作業(yè)前這個(gè)輸出目錄不應(yīng)該存在，否則Hadoop會(huì)報(bào)錯(cuò)并拒絕運(yùn)行該作業(yè)。這種預(yù)防措施的目的是防止數(shù)據(jù)丟失(如果長(zhǎng)時(shí)間運(yùn)行的數(shù)據(jù)結(jié)果被意外覆蓋，肯定是非常惱人的)。
l) System.exit(job.waitForCompletion(true) ? 0 : 1);job執(zhí)行，等待執(zhí)行結(jié)果

4. 各個(gè)包的功能

到此為止，三大部分就分析完畢，然后再來看看引入的有哪些類：

a） package org.apache.hadoop.examples;Java 提供包機(jī)制管理代碼，關(guān)鍵詞是package, 包名字可以自己定，但不能重復(fù)。通常為了包的唯一性，推薦使用公司域名的逆序作為包,于是有了上面例子中的‘org.apache.hadoop’這樣的包名。 
b) import java.io.IOException; 凡是以java開頭的包，在JDK1.7的API里可以找到類的資料。這里是從java.io中引入IOException,是一個(gè)輸入輸出異常類。 
c) import java.util.StringTokenizer;這是從java.util包中引入的StringTokenizer類，是一個(gè)解析文本的類。具體用法上文中已提過了。 
d) import org.apache.hadoop.conf.Configuration;凡是以org.apache.hadoop開頭的包，在Hadoop1.2.1 的API文檔可以找到類的資料。這里是從hadoop的conf包中引入Configuration類，它是一個(gè)讀寫和保存配置信息的類。
e) import org.apache.hadoop.fs.Path;  Path類保存文件或者目錄的路徑字符串 
f) import org.apache.hadoop.io.IntWritable;  IntWritable是一個(gè)以類表示的可序化的整數(shù)。在java中，要表示一個(gè)整數(shù)，可以使用int類型，也可以使用integer類型，integer封裝了int類型，且integer類是可序化的。但Hadoop認(rèn)為integer的可序化不合適，于是實(shí)現(xiàn)了IntWritable。 
g) import org.apache.hadoop.io.Text;  從io包中引入Text類，是一個(gè)存儲(chǔ)字符串的可比較可序化的類。 
h) import org.apache.hadoop.mapreduce.Job;  引入Job類，Hadoop中每個(gè)需要執(zhí)行的任務(wù)是一個(gè)Job,這個(gè)Job負(fù)責(zé)參數(shù)配置、設(shè)置MapReduce細(xì)節(jié)、提交到Hadoop集群、執(zhí)行控制等操作。 
i) import org.apache.hadoop.mapreduce.Mapper;引入Mapper類，負(fù)責(zé)MapReduce中的Map過程。 
j) import org.apache.hadoop.mapreduce.Reducer;引入Reduce類，負(fù)責(zé)MapReduce中的Reduce過程。 
k) import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;引入FileInputFormat類，主要功能是將文件進(jìn)行切片。
l) import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;FileOutputFormat類是將輸出結(jié)果寫入文件。
m) import org.apache.hadoop.util.GenericOptionsParser;這個(gè)類負(fù)責(zé)解析命令行參數(shù)。

從代碼的功能上，我們已經(jīng)對(duì)map reduce有了一個(gè)清晰的認(rèn)識(shí)，那么wordcount程序具體是怎么執(zhí)行的呢？

將文件file1.txt,file2.txt 上傳到hdfs中的hdfsinput1文件夾里（上傳的方式可以通過eclipse客戶端，也可以通過Hadoop命令行），然后在eclipse上編寫wordcount.java文件（也即是第一部分分析的源碼） 

由于測(cè)試用的文件較小，所以每個(gè)文件為一個(gè)split，并將文件按行分割形成< key,value>，這一步由MapReduce框架自動(dòng)完成，其中key值為該行首字母相對(duì)于文本文件首地址的偏移量。 

得到map方法輸出的< key,value>對(duì)后，進(jìn)行Combine操作。

同樣，在Reduce過程中先對(duì)輸入的數(shù)據(jù)進(jìn)行排序，再交由自定義的reduce方法進(jìn)行處理，得到新的< key,value>對(duì)，并作為WordCount的輸出結(jié)果，輸出結(jié)果存放在第一張圖的lxnoutputssss文件夾下的part-r-00000里。 

感謝各位的閱讀！關(guān)于“Hadoop怎么實(shí)現(xiàn)HelloWorld”這篇文章就分享到這里了，希望以上內(nèi)容可以對(duì)大家有一定的幫助，讓大家可以學(xué)到更多知識(shí)，如果覺得文章不錯(cuò)，可以把它分享出去讓更多的人看到吧！

本文名稱：Hadoop怎么實(shí)現(xiàn)HelloWorld
URL分享：http://m.jiaotiyi.com/article/iieooi.html