Hadoop多个文件单词计数

Hadoop的安装

首先下载Hadoop的安装包，这里使用2.7.3版本。解压到/usr/local下

sudo tar -zxvf hadoop-2.7.3.tar.gz -C /usr/local/ 

然后更改hadoop-2.7.3的属主

sudo chown -R jessin /usr/local/hadoop-2.7.3

在/etc/profile下添加HADOOP_HOME环境变量，并将其bin放到PATH路径下：

export HADOOP_HOME=/usr/local/hadoop-2.7.3
export PATH=${HADOOP_HOME}/bin:${HADOOP_HOME}/sbin:$PATH

接着使配置立即生效：

source /etc/profile

接下来在$HADOOP_HOME/etc/hadoop下修改或者添加四个文件，这时使用的是伪分布式，还是会使用HDFS。
core-site.xml

    
        hadoop.tmp.dir
        file:/usr/local/hadoop-2.7.3/tmp
        Abase for other temporary directories.
    

    
        fs.defaultFS
        hdfs://localhost:9000
    

hdfs-site.xml

    
        dfs.replication
        1
    
    
        dfs.namenode.name.dir
        file:/usr/local/hadoop-2.7.3/tmp/dfs/name
    
    
        dfs.datanode.data.dir
        file:/usr/local/hadoop-2.7.3/tmp/dfs/data
    

mapred-site.xml

    
        mapreduce.framework.name
        yarn
    

yarn-site.xml

    
        yarn.nodemanager.aux-services
        mapreduce_shuffle
    

必须在$HADOOP_HOME/etc/hadoop/hadoop-env.xml下添加JAVA_HOME的环境变量：

export JAVA_HOME=/usr/lib/jvm/jdk1.7.0

Hadoop的启动和终止

格式化HDFS

hdfs namenode -format

启动hdfs和yarn

start-dfs.sh
start-yarn.sh

在hdfs下建立一个HOME文件夹

hadoop fs -mkdir /user/jessin

结束Hadoop

stop-yarn.sh
stop-dfs.sh

单词计数程序

单词计数程序是Hadoop的hello world程序，这里使用Maven来构建，需要在pom.xml添加如下两个jar:

            org.apache.hadoop
            hadoop-common
            2.7.3
 
 
            org.apache.hadoop
            hadoop-mapreduce-client-core
            2.7.3
 

WordCount.java。需要注意的是Mapper和Reducer的泛型参数的前两个是输入的key,value类型，后两个是输出的key,value类型。

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 {

  //TokenizerMapper继承Mapper类，并重写其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 {
    //map方法中的value值存储的是文本文件中的一行信息（以回车符为行结束标记）。
    //map方法中的key值为该行的首字符相对与文本文件的首地址的偏移量。
    //StringTokenizer类将每一行拆分成一个个的单词，并将作为map方法的结果输出。
    //其中IntWritable和Text类是Hadoop对int和String类的序列化封装，这些类能够被序列化，以便在分布式环境中进行数据交换。
      StringTokenizer itr = new StringTokenizer(value.toString());
      System.out.println("key : " + key + " value : " + value);
      while (itr.hasMoreTokens()) {
        word.set(itr.nextToken());
        context.write(word, one);//输出为
      }
    }
  }
  //IntSumReducer继承Reducer类，并重写其reduce方法。
  public static class IntSumReducer 
       extends Reducer {
    private IntWritable result = new IntWritable();

    public void reduce(Text key, Iterable values, 
                       Context context
                       ) throws IOException, InterruptedException {
    //reduce方法的输入参数key为单个单词;
    //而Iterable values为各个Mapper上对应单词的计数值所组成的列表。
      int sum = 0;
      for (IntWritable val : values) {//遍历求和
        sum += val.get();
      }
      result.set(sum);
      context.write(key, result);//输出求和后的
    }
  }

  //在MapReduce中，由Job对象负责管理和运行一个计算任务，并通过Job的一些方法对任务的参数进行相关的设置。
  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);
    //使用TokenizerMapper类完成Map过程；
    job.setMapperClass(TokenizerMapper.class);
    //使用IntSumReducer类完成Combiner过程；
    job.setCombinerClass(IntSumReducer.class);
    //使用IntSumReducer类完成Reducer过程；
    job.setReducerClass(IntSumReducer.class);
    //设置了Map过程和Reduce过程的输出类型，其中设置key的输出类型为Text；
    job.setOutputKeyClass(Text.class);
    //设置了Map过程和Reduce过程的输出类型，其中设置value的输出类型为IntWritable；
    job.setOutputValueClass(IntWritable.class);
    //设置任务数据的输入路径；
    FileInputFormat.addInputPath(job, new Path(otherArgs[0]));
    //设置任务输出数据的保存路径；
    FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));
    //调用job.waitForCompletion(true) 执行任务，执行成功后退出；
    System.exit(job.waitForCompletion(true) ? 0 : 1);
  }
}

可以在根目录下打jar

mvn clean compile package

结果如下：

然后启动hadoop，在hdfs创建wordcount/input文件夹，将代码目录的的两个输入文件上传到hdfs

hadoop fs -mkdir -p wordcount/input
hadoop fs -put src/main/resources/input wordcount
hadoop fs -ls wordcount

结果如下：

在运行job之前，在HDFS上的输出文件夹必须不存在，否则会运行失败。可以使用以下命令删除HDFS上的文件夹。默认是用户的HOME目录：

hadoop fs -rm -r wordcount/output

使用以下命令将job提交到hadoop运行，注意最后两个是HDFS上的输入和输出文件，倒数第三个WordCount是jar中的主函数：

hadoop jar ~/Documents/Program/final/hadoop_helloworld/target/hadoop-helloworld-1.0-SNAPSHOT.jar WordCount wordcount/input wordcount/output

结果如下：

在hdfs的wordcount/output下生成了两个文件，其中part-r-00000含有输出结果

hadoop fs -ls wordcount/output
hadoop fs -cat wordcount/output/part-r-00000

运行结果如下：