MapReduce程序會(huì)根據(jù)輸入的文件產(chǎn)生多個(gè)map任務(wù)。Hadoop提供的Mapper類(lèi)是實(shí)現(xiàn)Map任務(wù)的一個(gè)抽象基類(lèi)，該基類(lèi)提供了一個(gè)map()方法，默認(rèn)情況下，Mapper類(lèi)中的map()方法是沒(méi)有做任何處理的。

如果我們想自定義map()方法，我們只需要繼承Mapper類(lèi)并重寫(xiě)map()方法即可。接下來(lái)，我們以詞頻統(tǒng)計(jì)為例，自定義一個(gè)map()方法，具體代碼如文件所示。

文件 WordCountMapper.java

 import java.io.IOException;

 import org.apache.hadoop.io.IntWritable;

 import org.apache.hadoop.io.LongWritable;

 import org.apache.hadoop.io.Text;

 import org.apache.hadoop.mapreduce.Mapper;

 public class WordCountMapper extends Mapper<LongWritable, Text,

       Text, IntWritable> {

   @Override

   protected void map(LongWritable key, Text value, Mapper<

        LongWritable, Text, Text, IntWritable>.Context context)

       throws IOException, InterruptedException {

     // 接收傳入進(jìn)來(lái)的一行文本，把數(shù)據(jù)類(lèi)型轉(zhuǎn)換為String類(lèi)型

     String line = value.toString();

     // 將這行內(nèi)容按照分隔符切割

     String[] words = line.split(" ");

     // 遍歷數(shù)組，每出現(xiàn)一個(gè)單詞就標(biāo)記一個(gè)數(shù)組1 例如：<單詞,1>

     for (String word : words) {

       // 使用context，把Map階段處理的數(shù)據(jù)發(fā)送給Reduce階段作為輸入數(shù)據(jù)

       **context.write(new Text(word), new IntWritable(1));**

     }

   }

 }

猜你喜歡：

Spark的集群安裝部署 

Spark有哪些特點(diǎn)，Spark的生態(tài)系統(tǒng)包含哪些組件?