阅读量:0
MapReduce是一种编程模型，用于处理和生成大数据集。在词频统计中，MapReduce通过映射（Map）阶段将文本拆分成单词并计算每个单词的出现次数，然后通过归约（Reduce）阶段合并所有单词的计数结果，从而得到最终的词频排序。
MapReduce 词频排序与统计
MapReduce 是一种广泛应用于大数据处理的编程模型，它通过将任务分解为多个小任务并行执行，从而提高数据处理的效率，在 Hadoop 生态系统中，MapReduce 被广泛用于大规模数据集的分析和处理，本文将详细介绍如何使用 MapReduce 实现词频统计及排序。
MapReduce 基本概念
1、Map 阶段：输入数据会被拆分成多个数据块（splits），每个数据块由一个 map 任务处理，Map 函数会接收一组键值对，并生成新的键值对作为中间结果，在词频统计中，输入的键值对是 \(<文本行号, 该行内容>\)，输出的键值对是 \(<单词, 1>\)。
2、Shuffle 和 Sort 阶段：Map 阶段的输出会经过 Shuffle 和 Sort 过程，将所有相同 key 的值聚合在一起，以便进行 reduce 操作。
3、Reduce 阶段：Reduce 函数会接收到一组键和对应的值列表，对这些值进行合并处理，生成最终的输出结果，在词频统计中，reduce 函数会计算每个单词的出现次数。
词频统计的 MapReduce 实现
Mapper 类
import org.apache.hadoop.io.IntWritable; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Mapper; import java.io.IOException; public class WordcountMapper extends Mapper<LongWritable, Text, Text, IntWritable> { private Text k = new Text(); private IntWritable v = new IntWritable(1); @Override protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { String line = value.toString(); String[] words = line.split("\\s+"); for (String word : words) { k.set(word); context.write(k, v); } }}Reducer 类
import org.apache.hadoop.io.IntWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Reducer; import java.io.IOException; public class WordcountReducer extends Reducer<Text, IntWritable, Text, IntWritable> { private IntWritable result = new IntWritable(); @Override protected 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); }}Driver 类
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 WordcountDriver { public static void main(String[] args) throws Exception { Configuration conf = new Configuration(); Job job = Job.getInstance(conf, "word count"); job.setJarByClass(WordcountDriver.class); job.setMapperClass(WordcountMapper.class); job.setCombinerClass(WordcountReducer.class); job.setReducerClass(WordcountReducer.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); }}词频排序的 MapReduce 实现
为了实现按词频降序排列，可以增加一个 MapReduce 任务来进行排序，具体步骤如下：
1、第二个 MapReduce 任务：第一个 MapReduce 任务的输出作为第二个 MapReduce 任务的输入，第二个任务负责对词频进行排序，可以通过设置自定义的排序比较器来实现。
自定义比较器类
import org.apache.hadoop.io.IntWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Partitioner; public class WordCountPartitioner extends Partitioner<Text, IntWritable> { @Override public int getPartition(Text key, IntWritable value, int numPartitions) { return (key.hashCode() & Integer.MAX_VALUE) % numPartitions; } }第二个 Reducer 类
import org.apache.hadoop.io.IntWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Reducer; import java.io.IOException; import java.util.Collections; import java.util.PriorityQueue; public class TopNReducer extends Reducer<Text, IntWritable, Text, IntWritable> { private PriorityQueue<WordCount> queue = new PriorityQueue<>(10, Collections.reverseOrder()); @Override protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException { int sum = 0; for (IntWritable val : values) { sum += val.get(); } WordCount wordCount = new WordCount(key, sum); if (queue.size() < 10) { queue.offer(wordCount); } else if (queue.peek().getCount() < wordCount.getCount()) { queue.poll(); queue.offer(wordCount); } } @Override protected void cleanup(Context context) throws IOException, InterruptedException { while (!queue.isEmpty()) { WordCount wordCount = queue.poll(); context.write(wordCount.getWord(), new IntWritable(wordCount.getCount())); } } }常见问题解答（FAQ）
1、问: MapReduce 中的 Shuffle 和 Sort 是什么？
答：Shuffle 和 Sort 是 MapReduce 框架中的两个重要过程，Shuffle 过程是指将 Map 任务的输出按照 key 进行分区，并传递给相应的 Reduce 任务，Sort 过程是指在 Shuffle 过程中，对每个分区内的数据按键进行排序，以便 Reduce 任务能够高效地处理这些数据，这两个过程共同确保了 Reduce 任务能够正确地接收到所有相关数据。
2、问: MapReduce 如何保证数据的完整性和正确性？
答：MapReduce 通过多种机制来保证数据的完整性和正确性，MapReduce 框架会自动处理输入数据的分片和备份，确保即使部分节点失败，整个任务也能完成，MapReduce 使用校验和等机制来验证数据的完整性，MapReduce 还支持任务重试和容错机制，确保任务在遇到错误时能够自动重试或恢复，用户可以通过编写自定义的 MapReduce 程序来控制数据的处理逻辑，从而进一步保证数据的正确性。
