文本中的常用词答案

【问题标题】：common word in a text文本中的常用词
【发布时间】：2014-01-26 00:08:54
【问题描述】：

我正在尝试使用 Hadoop 找出文本中最常见的单词。 Hadoop 是一个允许跨计算机集群分布式处理大型数据集的框架。

我知道这可以通过使用 Unix 命令轻松完成：job: sort -n -k2 txtname | tail。但这不适用于大型数据集。所以我试图分解问题，然后结合结果。

这是我的WordCount 课程：

    import java.util.Arrays;
    import org.apache.commons.lang.StringUtils;
    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.lib.input.FileInputFormat;
    import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

    public class WordCount {
      public static void runJob(String[] input, String output) throws Exception {
        Configuration conf = new Configuration();
        Job job = new Job(conf);
        job.setJarByClass(WordCount.class);

        job.setMapperClass(TokenizerMapper.class);
        job.setReducerClass(IntSumReducer.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(IntWritable.class);

        Path outputPath = new Path(output);
        FileInputFormat.setInputPaths(job, StringUtils.join(input, ","));
        FileOutputFormat.setOutputPath(job, outputPath);
        outputPath.getFileSystem(conf).delete(outputPath,true);
        job.waitForCompletion(true);
      }

      public static void main(String[] args) throws Exception {
        runJob(Arrays.copyOfRange(args, 0, args.length-1), args[args.length-1]);
      }
    }

我知道我需要做一个额外的工作来与 map reduce classes 并行工作。

这是我的TokenizerMapper 课程：

    import java.io.IOException;
    import java.util.StringTokenizer;
    import org.apache.hadoop.io.IntWritable;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.Mapper;

    public class TokenizerMapper extends Mapper<Object, Text, Text, IntWritable> { 
      private final IntWritable one = new IntWritable(1);
      private Text data = new Text();

      public void map(Object key, Text value, Context context) throws IOException, InterruptedException {
        StringTokenizer itr = new StringTokenizer(value.toString(), "-- \t\n\r\f,.:;?![]'\"");

        while (itr.hasMoreTokens())  {
          data.set(itr.nextToken().toLowerCase());
          context.write(data, one);
        }
      }
    }

这是我的IntSumReducer 课程：

    import java.io.IOException;
    import java.util.Iterator;
    import org.apache.hadoop.io.IntWritable;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.Reducer;

    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 value : values) {
          // TODO: complete code here
          sum+=value.get();
        }

        result.set(sum);

        // TODO: complete code here

        if (sum>3) {
          context.write(key,result);
        }
      }
    }

我需要做的是定义另一个 map 和 reduce 类，它们将与当前的类并行工作。出现次数最多的单词会出现，这是我目前所拥有的 reduce 类：

import java.io.IOException;
import java.util.Iterator;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.Reducer.Context;

public class reducer2 extends Reducer<Text, IntWritable, Text, IntWritable> {
  int max_sum =0;
  Text max_occured_key = new Text();

  private IntWritable result = new IntWritable();

  public void reduce(Text key, Iterable<IntWritable> values, Context context)

  throws IOException, InterruptedException {
    int sum = 0;

    for (IntWritable value : values) {
      // TODO: complete code here
      sum+=value.get();
    }

    if (sum >max_sum) {
      max_sum = sum;
      max_occured_key.set(key);
    }

    context.write(max_occured_key, new IntWritable(max_sum));

    //result.set(sum);

    // TODO: complete code here

    /*
    if (sum>3) {
      context.write(key,result);
    }
    */
  }

  protected void cleanup(Context context) throws IOException, InterruptedException {
    context.write(max_occured_key, new IntWritable(max_sum));
  }
}

mapper2的代码：

import java.io.IOException;
import java.util.StringTokenizer;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Mapper.Context;

public class mapper2 {
  private final IntWritable one = new IntWritable(1);
  private Text data = new Text();

  public void map(Object key, Text value, Context context) throws IOException, InterruptedException {
    StringTokenizer itr = new StringTokenizer(value.toString(), "-- \t\n\r\f,.:;?![]'\"");
    int count =0;

    while (itr.hasMoreTokens()) {    
      //data.set(itr.nextToken().toLowerCase());          
      context.write(data, one);
    }
  }
}

我还编辑了WordCount 类，以便可以同时运行两个作业：

import java.util.Arrays;
import org.apache.commons.lang.StringUtils;
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.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class WordCount {
  public static void runJob(String[] input, String output) throws Exception {
    Configuration conf = new Configuration();
    Job job = new Job(conf);
    job.setJarByClass(WordCount.class);

    job.setMapperClass(TokenizerMapper.class);
    job.setReducerClass(IntSumReducer.class);
    job.setMapOutputKeyClass(Text.class);
    job.setMapOutputValueClass(IntWritable.class);

    Path outputPath = new Path(output);
    FileInputFormat.setInputPaths(job, StringUtils.join(input, ","));
    FileOutputFormat.setOutputPath(job, outputPath);
    outputPath.getFileSystem(conf).delete(outputPath,true);
    job.waitForCompletion(true);

    Job job2 = new Job(conf);
    job2.setJarByClass(WordCount.class);

    job2.setMapperClass(TokenizerMapper.class);
    job2.setReducerClass(reducer2.class);
    job2.setMapOutputKeyClass(Text.class);
    job2.setMapOutputValueClass(IntWritable.class);

    Path outputPath2 = new Path(output);
    FileInputFormat.setInputPaths(job, StringUtils.join(input, ","));
    FileOutputFormat.setOutputPath(job, outputPath);
    outputPath.getFileSystem(conf).delete(outputPath,true);
    job.waitForCompletion(true);
  }

  public static void main(String[] args) throws Exception {
       runJob(Arrays.copyOfRange(args, 0, args.length-1), args[args.length-1]);
  }
}

如何使用 hadoop 找出文本中最常用的词？

【问题讨论】：

你有什么问题？
我正在尝试使用 hadoop 找出文本中最常见的单词并打印出来

标签： java hadoop

【解决方案1】：

这是典型的字数统计问题，您可以 google 并找到任意数量的基本字数统计解决方案。然后你只需要再做一步：返回计数最多的单词。

如何做到这一点？

如果数据量不是太大并且您可以负担使用单个 reducer，则将 reducer 的数量设置为 1。在您的 reduce 中保留一个局部变量，用于记住哪些组（即单词）有/有最高计数。然后将该结果写入 HDFS 中的文件。

如果数据量排除了使用单个 reducer 的可能性，那么您只需在上述第一个步骤之外多做一步：您需要在所有 reducer 中找到最高计数。您可以通过全局计数器或通过将每个单独的最大单词写入 hdfs 中自己的（小）文件并进行后处理步骤（可能是 linux 脚本）来解析并获得最大值的最大值。或者，您可以让另一个 map/reduce 作业找到它 - 但这对于那个小/简单的操作来说有点矫枉过正。

【讨论】：

【解决方案2】：

我认为您误认为sort -n -k2 在这种情况下无法大规模工作。 WordCount 可能永远不会输出数千或数万个单词。这仅仅是由于大多数自然语言的基数。因此，即使您有 10 PB 的数据，它仍然会被精简为最多 10,000 或 20,000 个单词（尽管数量很多）。

首先，让您的 WordCount 作业在数据上运行。然后，使用一些 bash 将前 N 个拉出来。

hadoop fs -cat /output/of/wordcount/part* | sort -n -k2 | tail -n20

如果由于某种原因，您从字数统计中得到了大量的单词（即，您不是在使用自然语言）...

有两个 MapReduce 作业：

WordCount：统计所有单词（几乎完全是示例）
TopN：一个 MapReduce 作业，用于查找某事物的前 N 个（以下是一些示例：source code、blog post）

让 WordCount 的输出写入 HDFS。然后，让 TopN 读取该输出。这称为作业链接，有多种方法可以解决此问题：oozie、bash 脚本、从驱动程序中触发两个作业等。

你需要两个工作的原因是你在做两个聚合：一个是字数，第二个是 topN。通常在 MapReduce 中，每个聚合都需要自己的 MapReduce 作业。

【讨论】：