可以通過(guò)一個(gè)簡(jiǎn)單的例子來(lái)說(shuō)明MapReduce到底是什么：

　　我們要統(tǒng)計(jì)一個(gè)大文件中的各個(gè)單詞出現(xiàn)的次數(shù)。由于文件太大。我們把這個(gè)文件切分成如果小文件，然后安排多個(gè)人去統(tǒng)計(jì)。這個(gè)過(guò)程就是”Map”。然后把每個(gè)人統(tǒng)計(jì)的數(shù)字合并起來(lái)，這個(gè)就是“Reduce"。

　　上面的例子如果在MapReduce去做呢，就需要?jiǎng)?chuàng)建一個(gè)任務(wù)job，由job把文件切分成若干獨(dú)立的數(shù)據(jù)塊，并分布在不同的機(jī)器節(jié)點(diǎn)中。然后通過(guò)分散在不同節(jié)點(diǎn)中的Map任務(wù)以完全并行的方式進(jìn)行處理。MapReduce會(huì)對(duì)Map的輸出地行收集，再將結(jié)果輸出送給Reduce進(jìn)行下一步的處理。

　　對(duì)于一個(gè)任務(wù)的具體執(zhí)行過(guò)程，會(huì)有一個(gè)名為"JobTracker"的進(jìn)程負(fù)責(zé)協(xié)調(diào)MapReduce執(zhí)行過(guò)程中的所有任務(wù)。若干條TaskTracker進(jìn)程用來(lái)運(yùn)行單獨(dú)的Map任務(wù)，并隨時(shí)將任務(wù)的執(zhí)行情況匯報(bào)給JobTracker。如果一個(gè)TaskTracker匯報(bào)任務(wù)失敗或者長(zhǎng)時(shí)間未對(duì)本身任務(wù)進(jìn)行匯報(bào)，JobTracker會(huì)啟動(dòng)另外一個(gè)TaskTracker重新執(zhí)行單獨(dú)的Map任務(wù)。

下面的具體的代碼實(shí)現(xiàn)：

1. 編寫(xiě)wordcount的相關(guān)job

(1)eclipse下創(chuàng)建相關(guān)maven項(xiàng)目，依賴(lài)jar包如下（也可參照hadoop源碼包下的hadoop-mapreduce-examples項(xiàng)目的pom配置）

　　注意：要配置一個(gè)maven插件maven-jar-plugin，并指定mainClass

<dependencies>
  <dependency>
   <groupId>junit</groupId>
   <artifactId>junit</artifactId>
   <version>4.11</version>
  </dependency>
  <dependency>
    <groupId>org.apache.hadoop</groupId>
    <artifactId>hadoop-mapreduce-client-core</artifactId>
    <version>2.5.2</version>
  </dependency>
  <dependency>
    <groupId>org.apache.hadoop</groupId>
    <artifactId>hadoop-common</artifactId>
    <version>2.5.2</version>
  </dependency>
 </dependencies>
 
 <build>
   <plugins>
     <plugin>
  <groupId>org.apache.maven.plugins</groupId>
   <artifactId>maven-jar-plugin</artifactId>
   <configuration>
    <archive>
     <manifest>
      <mainClass>com.xxx.demo.hadoop.wordcount.WordCount</mainClass>
     </manifest>
    </archive>
   </configuration>
  </plugin>
   </plugins>
 </build>

(2)根據(jù)MapReduce的運(yùn)行機(jī)制，一個(gè)job至少要編寫(xiě)三個(gè)類(lèi)分別用來(lái)完成Map邏輯、Reduce邏輯、作業(yè)調(diào)度這三件事。

Map的代碼可繼承org.apache.hadoop.mapreduce.Mapper類(lèi)

public static class TokenizerMapper
    extends Mapper<Object, Text, Text, IntWritable>{
 
  private final static IntWritable one = new IntWritable(1);
  private Text word = new Text();
   //由于該例子未用到key的參數(shù)，所以該處key的類(lèi)型就簡(jiǎn)單指定為Object
  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);
   }
  }
 }

Reduce的代碼可繼承org.apache.hadoop.mapreduce.Reducer類(lèi)

public class IntSumReducer
    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();
   }
   result.set(sum);
   context.write(key, result);
  }
 }

編寫(xiě)main方法進(jìn)行作業(yè)調(diào)度

public static void main(String[] args) throws Exception {
  Configuration conf = new Configuration();
  Job job = Job.getInstance(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(args[0]));
  FileOutputFormat.setOutputPath(job, new Path(args[1]));
  job.waitForCompletion(true) ;
  //System.exit(job.waitForCompletion(true) ? 0 : 1);
 }

2. 上傳數(shù)據(jù)文件到hadoop集群環(huán)境

執(zhí)行mvn install把項(xiàng)目打成jar文件然后上傳到linux集群環(huán)境，使用hdfs dfs -mkdir命令在hdfs文件系統(tǒng)中創(chuàng)建相應(yīng)的命令，使用hdfs dfs -put 把需要處理的數(shù)據(jù)文件上傳到hdfs系統(tǒng)中，示例：hdfs dfs -put ${linux_path/數(shù)據(jù)文件} ${hdfs_path}

3. 執(zhí)行job

在集群環(huán)境中執(zhí)行命令: hadoop jar ${linux_path}/wordcount.jar ${hdfs_input_path} ${hdfs_output_path}

4. 查看統(tǒng)計(jì)結(jié)果

hdfs dfs -cat ${hdfs_output_path}/輸出文件名

以上的方式在未啟動(dòng)hadoop集群環(huán)境時(shí)，是以Local模式運(yùn)行，此時(shí)HDFS和YARN都不起作用。下面是在偽分布式模式下執(zhí)行mapreduce job時(shí)需要做的工作，先把官網(wǎng)上列的步驟摘錄出來(lái)：

配置主機(jī)名

# vi /etc/sysconfig/network

例如：

NETWORKING=yes
HOSTNAME=master


vi /etc/hosts

填入以下內(nèi)容

127.0.0.1 localhost

配置ssh免密碼互通

ssh-keygen -t rsa
# cat?~/.ssh/id_rsa.pub?>>?~/.ssh/authorized_keys

配置core-site.xml文件（位于${HADOOP_HOME}/etc/hadoop/

<configuration>
  <property>
    <name>fs.defaultFS</name>
    <value>hdfs://localhost:9000</value>
  </property>
</configuration>

配置hdfs-site.xml文件

<configuration>
  <property>
    <name>dfs.replication</name>
    <value>1</value>
  </property>
</configuration>

下面的命令可以在單機(jī)偽分布模式下運(yùn)行mapreduce的job

1.Format the filesystem:
$ bin/hdfs namenode -format
2.Start NameNode daemon and DataNode daemon:
$ sbin/start-dfs.sh
3.The hadoop daemon log output is written to the $HADOOP_LOG_DIR directory (defaults to $HADOOP_HOME/logs).

4.Browse the web interface for the NameNode; by default it is available at:
NameNode - http://localhost:50070/
Make the HDFS directories required to execute MapReduce jobs:
$ bin/hdfs dfs -mkdir /user
$ bin/hdfs dfs -mkdir /user/<username>
5.Copy the input files into the distributed filesystem:
$ bin/hdfs dfs -put etc/hadoop input
6.Run some of the examples provided:
$ bin/hadoop jar share/hadoop/mapreduce/hadoop-mapreduce-examples-2.5.2.jar grep input output 'dfs[a-z.]+'
7.Examine the output files:
Copy the output files from the distributed filesystem to the local filesystem and examine them:

$ bin/hdfs dfs -get output output
$ cat output/*
or

View the output files on the distributed filesystem:

$ bin/hdfs dfs -cat output/*
8.When you're done, stop the daemons with:
$ sbin/stop-dfs.sh

總結(jié)

以上就是本文關(guān)于hadoop的wordcount實(shí)例代碼的全部?jī)?nèi)容，希望對(duì)大家有所幫助。感興趣的朋友可以繼續(xù)參閱本站其他相關(guān)專(zhuān)題，如有不足之處，歡迎留言指出。感謝朋友們對(duì)本站的支持！

最新評(píng)論