加勒比AV一二三区,浪潮精品无码久久久久久久久,亚洲电影第一页av中文字幕

新聞中心

這里有您想知道的互聯(lián)網(wǎng)營銷解決方案

Hadoop集群系列7：WordCount運(yùn)行詳解

1、MapReduce理論簡介

成都創(chuàng)新互聯(lián)公司是創(chuàng)新、創(chuàng)意、研發(fā)型一體的綜合型網(wǎng)站建設(shè)公司，自成立以來公司不斷探索創(chuàng)新，始終堅(jiān)持為客戶提供滿意周到的服務(wù)，在本地打下了良好的口碑，在過去的十載時間我們累計(jì)服務(wù)了上千家以及全國政企客戶，如成都石涼亭等企業(yè)單位，完善的項(xiàng)目管理流程，嚴(yán)格把控項(xiàng)目進(jìn)度與質(zhì)量監(jiān)控加上過硬的技術(shù)實(shí)力獲得客戶的一致表揚(yáng)。

1.1 MapReduce編程模型

　　MapReduce采用"分而治之"的思想，把對大規(guī)模數(shù)據(jù)集的操作，分發(fā)給一個主節(jié)點(diǎn)管理下的各個分節(jié)點(diǎn)共同完成，然后通過整合各個節(jié)點(diǎn)的中間結(jié)果，得到最終結(jié)果。簡單地說，MapReduce就是"任務(wù)的分解與結(jié)果的匯總"。

　　在Hadoop中，用于執(zhí)行MapReduce任務(wù)的機(jī)器角色有兩個：一個是JobTracker；另一個是TaskTracker，JobTracker是用于調(diào)度工作的，TaskTracker是用于執(zhí)行工作的。一個Hadoop集群中只有一臺JobTracker。

　　在分布式計(jì)算中，MapReduce框架負(fù)責(zé)處理了并行編程中分布式存儲、工作調(diào)度、負(fù)載均衡、容錯均衡、容錯處理以及網(wǎng)絡(luò)通信等復(fù)雜問題，把處理過程高度抽象為兩個函數(shù)：map和reduce，map負(fù)責(zé)把任務(wù)分解成多個任務(wù)，reduce負(fù)責(zé)把分解后多任務(wù)處理的結(jié)果匯總起來。

　　需要注意的是，用MapReduce來處理的數(shù)據(jù)集（或任務(wù)）必須具備這樣的特點(diǎn)：待處理的數(shù)據(jù)集可以分解成許多小的數(shù)據(jù)集，而且每一個小數(shù)據(jù)集都可以完全并行地進(jìn)行處理。

1.2 MapReduce處理過程

　　在Hadoop中，每個MapReduce任務(wù)都被初始化為一個Job，每個Job又可以分為兩種階段：map階段和reduce階段。這兩個階段分別用兩個函數(shù)表示，即map函數(shù)和reduce函數(shù)。map函數(shù)接收一個形式的輸入，然后同樣產(chǎn)生一個形式的中間輸出，Hadoop函數(shù)接收一個如形式的輸入，然后對這個value集合進(jìn)行處理，每個reduce產(chǎn)生0或1個輸出，reduce的輸出也是形式的。

MapReduce處理大數(shù)據(jù)集的過程

2、運(yùn)行WordCount程序

　　單詞計(jì)數(shù)是最簡單也是最能體現(xiàn)MapReduce思想的程序之一，可以稱為MapReduce版"Hello World"，該程序的完整代碼可以在Hadoop安裝包的"src/examples"目錄下找到。單詞計(jì)數(shù)主要完成功能是：統(tǒng)計(jì)一系列文本文件中每個單詞出現(xiàn)的次數(shù)，如下圖所示。

2.1 準(zhǔn)備工作

　　現(xiàn)在以"hadoop"普通用戶登錄"Master.Hadoop"服務(wù)器。

　　1）創(chuàng)建本地示例文件

　　首先在"/home/hadoop"目錄下創(chuàng)建文件夾"file"。

　　接著創(chuàng)建兩個文本文件file1.txt和file2.txt，使file1.txt內(nèi)容為"Hello World"，而file2.txt的內(nèi)容為"Hello Hadoop"。

　　2）在HDFS上創(chuàng)建輸入文件夾

　　3）上傳本地file中文件到集群的input目錄下

2.2 運(yùn)行例子

　　1）在集群上運(yùn)行WordCount程序

　　備注：以input作為輸入目錄，output目錄作為輸出目錄。

　　已經(jīng)編譯好的WordCount的Jar在"/usr/hadoop"下面，就是"hadoop-examples-1.0.0.jar"，所以在下面執(zhí)行命令時記得把路徑寫全了，不然會提示找不到該Jar包。

　　2）MapReduce執(zhí)行過程顯示信息

　　Hadoop命令會啟動一個JVM來運(yùn)行這個MapReduce程序，并自動獲得Hadoop的配置，同時把類的路徑（及其依賴關(guān)系）加入到Hadoop的庫中。以上就是Hadoop Job的運(yùn)行記錄，從這里可以看到，這個Job被賦予了一個ID號：job_201202292213_0002，而且得知輸入文件有兩個（Total input paths to process : 2），同時還可以了解map的輸入輸出記錄（record數(shù)及字節(jié)數(shù)），以及reduce輸入輸出記錄。比如說，在本例中，map的task數(shù)量是2個，reduce的task數(shù)量是一個。map的輸入record數(shù)是2個，輸出record數(shù)是4個等信息。

2.3 查看結(jié)果

　　1）查看HDFS上output目錄內(nèi)容

　　從上圖中知道生成了三個文件，我們的結(jié)果在"part-r-00000"中。

　　2）查看結(jié)果輸出文件內(nèi)容

#p#

3、WordCount源碼分析

3.1 特別數(shù)據(jù)類型介紹

　　Hadoop提供了如下內(nèi)容的數(shù)據(jù)類型，這些數(shù)據(jù)類型都實(shí)現(xiàn)了WritableComparable接口，以便用這些類型定義的數(shù)據(jù)可以被序列化進(jìn)行網(wǎng)絡(luò)傳輸和文件存儲，以及進(jìn)行大小比較。

BooleanWritable：標(biāo)準(zhǔn)布爾型數(shù)值

ByteWritable：單字節(jié)數(shù)值

DoubleWritable：雙字節(jié)數(shù)

FloatWritable：浮點(diǎn)數(shù)

IntWritable：整型數(shù)

LongWritable：長整型數(shù)

Text：使用UTF8格式存儲的文本

NullWritable：當(dāng)中的key或value為空時使用

3.2 舊的WordCount分析

　　1）源代碼程序

 
 
 
 
  
  
  
  package org.apache.hadoop.examples; 
  
  
  
   
  
  
  
  import java.io.IOException; 
  
  
  
  import java.util.Iterator; 
  
  
  
  import java.util.StringTokenizer; 
  
  
  
   
  
  
  
  import org.apache.hadoop.fs.Path; 
  
  
  
  import org.apache.hadoop.io.IntWritable; 
  
  
  
  import org.apache.hadoop.io.LongWritable; 
  
  
  
  import org.apache.hadoop.io.Text; 
  
  
  
  import org.apache.hadoop.mapred.FileInputFormat; 
  
  
  
  import org.apache.hadoop.mapred.FileOutputFormat; 
  
  
  
  import org.apache.hadoop.mapred.JobClient; 
  
  
  
  import org.apache.hadoop.mapred.JobConf; 
  
  
  
  import org.apache.hadoop.mapred.MapReduceBase; 
  
  
  
  import org.apache.hadoop.mapred.Mapper; 
  
  
  
  import org.apache.hadoop.mapred.OutputCollector; 
  
  
  
  import org.apache.hadoop.mapred.Reducer; 
  
  
  
  import org.apache.hadoop.mapred.Reporter; 
  
  
  
  import org.apache.hadoop.mapred.TextInputFormat; 
  
  
  
  import org.apache.hadoop.mapred.TextOutputFormat; 
  
  
  
   
  
  
  
  public class WordCount { 
  
  
  
   
  
  
  
      public static class Map extends MapReduceBase implements 
  
  
  
              Mapper { 
  
  
  
          private final static IntWritable one = new IntWritable(1); 
  
  
  
          private Text word = new Text(); 
  
  
  
   
  
  
  
          public void map(LongWritable key, Text value, 
  
  
  
                  OutputCollector output, Reporter reporter) 
  
  
  
                  throws IOException { 
  
  
  
              String line = value.toString(); 
  
  
  
              StringTokenizer tokenizer = new StringTokenizer(line); 
  
  
  
              while (tokenizer.hasMoreTokens()) { 
  
  
  
                  word.set(tokenizer.nextToken()); 
  
  
  
                  output.collect(word, one); 
  
  
  
              } 
  
  
  
          } 
  
  
  
      } 
  
  
  
   
  
  
  
      public static class Reduce extends MapReduceBase implements 
  
  
  
              Reducer { 
  
  
  
          public void reduce(Text key, Iterator values, 
  
  
  
                  OutputCollector output, Reporter reporter) 
  
  
  
                  throws IOException { 
  
  
  
              int sum = 0; 
  
  
  
              while (values.hasNext()) { 
  
  
  
                  sum += values.next().get(); 
  
  
  
              } 
  
  
  
              output.collect(key, new IntWritable(sum)); 
  
  
  
          } 
  
  
  
      } 
  
  
  
   
  
  
  
      public static void main(String[] args) throws Exception { 
  
  
  
          JobConf conf = new JobConf(WordCount.class); 
  
  
  
          conf.setJobName("wordcount"); 
  
  
  
   
  
  
  
          conf.setOutputKeyClass(Text.class); 
  
  
  
          conf.setOutputValueClass(IntWritable.class); 
  
  
  
   
  
  
  
          conf.setMapperClass(Map.class); 
  
  
  
          conf.setCombinerClass(Reduce.class); 
  
  
  
          conf.setReducerClass(Reduce.class); 
  
  
  
   
  
  
  
          conf.setInputFormat(TextInputFormat.class); 
  
  
  
          conf.setOutputFormat(TextOutputFormat.class); 
  
  
  
   
  
  
  
          FileInputFormat.setInputPaths(conf, new Path(args[0])); 
  
  
  
          FileOutputFormat.setOutputPath(conf, new Path(args[1])); 
  
  
  
   
  
  
  
          JobClient.runJob(conf); 
  
  
  
      } 
  
  
  
  }

　　3）主方法Main分析

  
  
  
  
   
   
   
   public static void main(String[] args) throws Exception { 
   
   
   
       JobConf conf = new JobConf(WordCount.class); 
   
   
   
       conf.setJobName("wordcount"); 
   
   
   
    
   
   
   
       conf.setOutputKeyClass(Text.class); 
   
   
   
       conf.setOutputValueClass(IntWritable.class); 
   
   
   
    
   
   
   
       conf.setMapperClass(Map.class); 
   
   
   
       conf.setCombinerClass(Reduce.class); 
   
   
   
       conf.setReducerClass(Reduce.class); 
   
   
   
    
   
   
   
       conf.setInputFormat(TextInputFormat.class); 
   
   
   
       conf.setOutputFormat(TextOutputFormat.class); 
   
   
   
    
   
   
   
       FileInputFormat.setInputPaths(conf, new Path(args[0])); 
   
   
   
       FileOutputFormat.setOutputPath(conf, new Path(args[1])); 
   
   
   
    
   
   
   
       JobClient.runJob(conf); 
   
   
   
   }

　　首先講解一下Job的初始化過程。main函數(shù)調(diào)用Jobconf類來對MapReduce Job進(jìn)行初始化，然后調(diào)用setJobName()方法命名這個Job。對Job進(jìn)行合理的命名有助于更快地找到Job，以便在JobTracker和Tasktracker的頁面中對其進(jìn)行監(jiān)視。

JobConf conf = new JobConf(WordCount. class ); conf.setJobName("wordcount" );

　　接著設(shè)置Job輸出結(jié)果的中key和value數(shù)據(jù)類型，因?yàn)榻Y(jié)果是<單詞,個數(shù)>，所以key設(shè)置為"Text"類型，相當(dāng)于Java中String類型。Value設(shè)置為"IntWritable"，相當(dāng)于Java中的int類型。

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

　　然后設(shè)置Job處理的Map（拆分）、Combiner（中間結(jié)果合并）以及Reduce（合并）的相關(guān)處理類。這里用Reduce類來進(jìn)行Map產(chǎn)生的中間結(jié)果合并，避免給網(wǎng)絡(luò)數(shù)據(jù)傳輸產(chǎn)生壓力。

conf.setMapperClass(Map.class );
conf.setCombinerClass(Reduce.class );
conf.setReducerClass(Reduce.class );

　　接著就是調(diào)用setInputPath()和setOutputPath()設(shè)置輸入輸出路徑。

conf.setInputFormat(TextInputFormat.class );
conf.setOutputFormat(TextOutputFormat.class );

　　（1）InputFormat和InputSplit

　　InputSplit是Hadoop定義的用來傳送給每個單獨(dú)的map的數(shù)據(jù)，InputSplit存儲的并非數(shù)據(jù)本身，而是一個分片長度和一個記錄數(shù)據(jù)位置的數(shù)組。生成InputSplit的方法可以通過InputFormat()來設(shè)置。

　　當(dāng)數(shù)據(jù)傳送給map時，map會將輸入分片傳送到InputFormat，InputFormat則調(diào)用方法getRecordReader()生成RecordReader，RecordReader再通過creatKey()、creatValue()方法創(chuàng)建可供map處理的對。簡而言之，InputFormat()方法是用來生成可供map處理的對的。

　　Hadoop預(yù)定義了多種方法將不同類型的輸入數(shù)據(jù)轉(zhuǎn)化為map能夠處理的對，它們都繼承自InputFormat，分別是：

  
  
  
  
   
   
   
   InputFormat 
   
   
   
    
   
   
   
       | 
   
   
   
    
   
   
   
       |---BaileyBorweinPlouffe.BbpInputFormat 
   
   
   
    
   
   
   
       |---ComposableInputFormat 
   
   
   
    
   
   
   
       |---CompositeInputFormat 
   
   
   
    
   
   
   
       |---DBInputFormat 
   
   
   
    
   
   
   
       |---DistSum.Machine.AbstractInputFormat 
   
   
   
    
   
   
   
       |---FileInputFormat 
   
   
   
    
   
   
   
           |---CombineFileInputFormat 
   
   
   
    
   
   
   
           |---KeyValueTextInputFormat 
   
   
   
    
   
   
   
           |---NLineInputFormat 
   
   
   
    
   
   
   
           |---SequenceFileInputFormat 
   
   
   
    
   
   
   
           |---TeraInputFormat 
   
   
   
    
   
   
   
           |---TextInputFormat

　　其中TextInputFormat是Hadoop默認(rèn)的輸入方法，在TextInputFormat中，每個文件（或其一部分）都會單獨(dú)地作為map的輸入，而這個是繼承自FileInputFormat的。之后，每行數(shù)據(jù)都會生成一條記錄，每條記錄則表示成形式：

key值是每個數(shù)據(jù)的記錄在數(shù)據(jù)分片中字節(jié)偏移量，數(shù)據(jù)類型是LongWritable；

value值是每行的內(nèi)容，數(shù)據(jù)類型是Text。

　?。?）OutputFormat

　　每一種輸入格式都有一種輸出格式與其對應(yīng)。默認(rèn)的輸出格式是TextOutputFormat，這種輸出方式與輸入類似，會將每條記錄以一行的形式存入文本文件。不過，它的鍵和值可以是任意形式的，因?yàn)槌绦騼?nèi)容會調(diào)用toString()方法將鍵和值轉(zhuǎn)換為String類型再輸出。

　　3）Map類中map方法分析

  
  
  
  
   
   
   
   public static class Map extends MapReduceBase implements 
   
   
   
           Mapper { 
   
   
   
       private final static IntWritable one = new IntWritable(1); 
   
   
   
       private Text word = new Text(); 
   
   
   
    
   
   
   
       public void map(LongWritable key, Text value, 
   
   
   
               OutputCollector output, Reporter reporter) 
   
   
   
               throws IOException { 
   
   
   
           String line = value.toString(); 
   
   
   
           StringTokenizer tokenizer = new StringTokenizer(line); 
   
   
   
           while (tokenizer.hasMoreTokens()) { 
   
   
   
               word.set(tokenizer.nextToken()); 
   
   
   
               output.collect(word, one); 
   
   
   
           } 
   
   
   
       } 
   
   
   
   }

　　Map類繼承自MapReduceBase，并且它實(shí)現(xiàn)了Mapper接口，此接口是一個規(guī)范類型，它有4種形式的參數(shù)，分別用來指定map的輸入key值類型、輸入value值類型、輸出key值類型和輸出value值類型。在本例中，因?yàn)槭褂玫氖荰extInputFormat，它的輸出key值是LongWritable類型，輸出value值是Text類型，所以map的輸入類型為。在本例中需要輸出這樣的形式，因此輸出的key值類型是Text，輸出的value值類型是IntWritable。

　　實(shí)現(xiàn)此接口類還需要實(shí)現(xiàn)map方法，map方法會具體負(fù)責(zé)對輸入進(jìn)行操作，在本例中，map方法對輸入的行以空格為單位進(jìn)行切分，然后使用OutputCollect收集輸出的。

　　4）Reduce類中reduce方法分析

  
  
  
  
   
   
   
   public static class Reduce extends MapReduceBase implements 
   
   
   
           Reducer { 
   
   
   
       public void reduce(Text key, Iterator values, 
   
   
   
               OutputCollector output, Reporter reporter) 
   
   
   
               throws IOException { 
   
   
   
           int sum = 0; 
   
   
   
           while (values.hasNext()) { 
   
   
   
               sum += values.next().get(); 
   
   
   
           } 
   
   
   
           output.collect(key, new IntWritable(sum)); 
   
   
   
       } 
   
   
   
   }

　　Reduce類也是繼承自MapReduceBase的，需要實(shí)現(xiàn)Reducer接口。Reduce類以map的輸出作為輸入，因此Reduce的輸入類型是。而Reduce的輸出是單詞和它的數(shù)目，因此，它的輸出類型是。Reduce類也要實(shí)現(xiàn)reduce方法，在此方法中，reduce函數(shù)將輸入的key值作為輸出的key值，然后將獲得多個value值加起來，作為輸出的值。

3.3 新的WordCount分析

　　1）源代碼程序

  
  
  
  
   
   
   
   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(); 
   
   
   
   　　　　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); 
   
   
   
   } 
   
   
   
   }

　 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);
　　}
}

　　Map過程需要繼承org.apache.hadoop.mapreduce包中Mapper類，并重寫其map方法。通過在map方法中添加兩句把key值和value值輸出到控制臺的代碼，可以發(fā)現(xiàn)map方法中value值存儲的是文本文件中的一行（以回車符為行結(jié)束標(biāo)記），而key值為該行的首字母相對于文本文件的首地址的偏移量。然后StringTokenizer類將每一行拆分成為一個個的單詞，并將作為map方法的結(jié)果輸出，其余的工作都交有MapReduce框架處理。

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);
　　}
}

　　Reduce過程需要繼承org.apache.hadoop.mapreduce包中Reducer類，并重寫其reduce方法。Map過程輸出中key為單個單詞，而values是對應(yīng)單詞的計(jì)數(shù)值所組成的列表，Map的輸出就是Reduce的輸入，所以reduce方法只要遍歷values并求和，即可得到某個單詞的總次數(shù)。

3）執(zhí)行MapReduce任務(wù)

  
  
  
  
   
   
   
   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  "); 
   
   
   
   　　　　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); 
   
   
   
   }

　　在MapReduce中，由Job對象負(fù)責(zé)管理和運(yùn)行一個計(jì)算任務(wù)，并通過Job的一些方法對任務(wù)的參數(shù)進(jìn)行相關(guān)的設(shè)置。此處設(shè)置了使用TokenizerMapper完成Map過程中的處理和使用IntSumReducer完成Combine和Reduce過程中的處理。還設(shè)置了Map過程和Reduce過程的輸出類型：key的類型為Text，value的類型為IntWritable。任務(wù)的輸出和輸入路徑則由命令行參數(shù)指定，并由FileInputFormat和FileOutputFormat分別設(shè)定。完成相應(yīng)任務(wù)的參數(shù)設(shè)定后，即可調(diào)用job.waitForCompletion()方法執(zhí)行任務(wù)。

4、WordCount處理過程

　　本節(jié)將對WordCount進(jìn)行更詳細(xì)的講解。詳細(xì)執(zhí)行步驟如下：

　　1）將文件拆分成splits，由于測試用的文件較小，所以每個文件為一個split，并將文件按行分割形成對，如圖4-1所示。這一步由MapReduce框架自動完成，其中偏移量（即key值）包括了回車所占的字符數(shù)（Windows和Linux環(huán)境會不同）。

圖4-1 分割過程

　　2）將分割好的對交給用戶定義的map方法進(jìn)行處理，生成新的對，如圖4-2所示。

圖4-2 執(zhí)行map方法

　　3）得到map方法輸出的對后，Mapper會將它們按照key值進(jìn)行排序，并執(zhí)行Combine過程，將key至相同value值累加，得到Mapper的最終輸出結(jié)果。如圖4-3所示。

圖4-3 Map端排序及Combine過程

　　4）Reducer先對從Mapper接收的數(shù)據(jù)進(jìn)行排序，再交由用戶自定義的reduce方法進(jìn)行處理，得到新的對，并作為WordCount的輸出結(jié)果，如圖4-4所示。

圖4-4 Reduce端排序及輸出結(jié)果

5、MapReduce新舊改變

　　Hadoop最新版本的MapReduce Release 0.20.0的API包括了一個全新的Mapreduce JAVA API，有時候也稱為上下文對象。

　　新的API類型上不兼容以前的API，所以，以前的應(yīng)用程序需要重寫才能使新的API發(fā)揮其作用。

　　新的API和舊的API之間有下面幾個明顯的區(qū)別。

新的API傾向于使用抽象類，而不是接口，因?yàn)檫@更容易擴(kuò)展。例如，你可以添加一個方法(用默認(rèn)的實(shí)現(xiàn))到一個抽象類而不需修改類之前的實(shí)現(xiàn)方法。在新的API中，Mapper和Reducer是抽象類。

新的API是在org.apache.hadoop.mapreduce包(和子包)中的。之前版本的API則是放在org.apache.hadoop.mapred中的。

新的API廣泛使用context object(上下文對象)，并允許用戶代碼與MapReduce系統(tǒng)進(jìn)行通信。例如，MapContext基本上充當(dāng)著JobConf的OutputCollector和Reporter的角色。

新的API同時支持"推"和"拉"式的迭代。在這兩個新老API中，鍵/值記錄對被推mapper中，但除此之外，新的API允許把記錄從map()方法中拉出，這也適用于reducer。"拉"式的一個有用的例子是分批處理記錄，而不是一個接一個。

新的API統(tǒng)一了配置。舊的API有一個特殊的JobConf對象用于作業(yè)配置，這是一個對于Hadoop通常的Configuration對象的擴(kuò)展。在新的API中，這種區(qū)別沒有了，所以作業(yè)配置通過Configuration來完成。作業(yè)控制的執(zhí)行由Job類來負(fù)責(zé)，而不是JobClient，它在新的API中已經(jīng)蕩然無存。

原文鏈接：http://www.cnblogs.com/xia520pi/archive/2012/05/16/2504205.html

【編輯推薦】

Hadoop集群系列1:CentOS安裝配置
Hadoop集群系列2:機(jī)器信息分布表
Hadoop集群系列3:VSFTP安裝配置
Hadoop集群系列4:SecureCRT使用
Hadoop集群搭建過程中相關(guān)環(huán)境配置詳解
Hadoop完全分布模式安裝實(shí)現(xiàn)詳解

本文題目：Hadoop集群系列7：WordCount運(yùn)行詳解
鏈接地址：http://www.dlmjj.cn/article/cdieide.html

日本综合一区二区|亚洲中文天堂综合|日韩欧美自拍一区|男女精品天堂一区|欧美自拍第6页亚洲成人精品一区|亚洲黄色天堂一区二区成人|超碰91偷拍第一页|日韩av夜夜嗨中文字幕|久久蜜综合视频官网|精美人妻一区二区三区

新聞中心

其他資訊