在 Cloudera Distribution Hadoop(CDH) 中执行 WordCount 程序

Cloudera Distribution Hadoop(CDH) 是业界领先的大数据处理平台之一，它提供了各种组件和工具，包括 HDFS（分布式文件系统）、MapReduce、Hive、Pig 等等，可以帮助我们轻松地处理大规模数据的存储和计算。在本文中，我们将介绍如何使用 Cloudera Distribution Hadoop(CDH) 在 MapReduce 中执行 WordCount 程序。

步骤一：准备数据

首先，我们需要准备一份需要处理的“数据”。在本例中，我们将使用一个包含若干英文单词的文本文件作为输入数据。可以从网络上下载这样的数据文件，或者自己编写一个简单的脚本来生成。

步骤二：编写 WordCount 程序

WordCount 程序是一个经典的 MapReduce 示例程序，它的作用是统计输入数据中各单词出现的次数。下面是一个简单的 WordCount 程序的代码：

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;

public class WordCount {
  
  public static class TokenizerMapper
       extends Mapper<Object, Text, Text, IntWritable>{
    
    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<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);
    }
  }

  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]));
    System.exit(job.waitForCompletion(true) ? 0 : 1);
  }
}

该程序的主要逻辑分为 Mapper 和 Reducer 两个部分。在 Mapper 部分，我们首先将输入数据按空格进行分词，然后将每个单词作为输出的键，把值设置为固定的 1。在 Reducer 部分，我们对输入的键值对进行合并操作，统计每个单词出现的次数。

步骤三：打包程序并上传到 Hadoop 集群

在运行 MapReduce 程序之前，我们需要将程序打包成一个 jar 文件，并上传到 Hadoop 集群中。可以使用以下命令来打包程序：

javac -classpath $(hadoop classpath) WordCount.java
jar cvf WordCount.jar *.class

其中，$(hadoop classpath) 可以获取 Hadoop 的 classpath，确保编译程序时可以引用到 Hadoop 的相关类库。

步骤四：运行 MapReduce 程序并查看结果

在打包好程序后，我们可以使用以下命令来在 Hadoop 集群上运行 MapReduce 程序：

hadoop jar WordCount.jar WordCount /input /output

其中，WordCount 是程序的主类名称，/input 是输入数据的路径，/output 是输出结果的路径。

运行程序后，可以通过以下命令来查看输出结果：

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

输出结果将显示为 key-value 对的形式，其中 key 表示单词，value 表示该单词在输入数据中出现的次数。

总结

通过本文的介绍，我们了解了如何使用 Cloudera Distribution Hadoop(CDH) 在 MapReduce 中执行 WordCount 程序。实际上，除了 WordCount，还有很多其他的 MapReduce 示例程序，可以帮助我们熟悉 Hadoop 的使用和 MapReduce 的编程模型。在实际开发中，我们可以根据需求编写自己的 MapReduce 程序，并利用 Hadoop 集群进行大规模数据处理。