【详解】HadoopMapReduce实战

举报
皮牙子抓饭 发表于 2025/10/10 21:57:43 2025/10/10
【摘要】 Hadoop MapReduce实战前言在大数据处理领域，Hadoop是一个非常重要的开源框架，它能够支持在廉价的硬件上运行大型分布式数据处理应用。Hadoop的核心组件之一是MapReduce，这是一种编程模型，用于大规模数据集（大于1TB）的并行处理。本文将通过一个具体的例子来介绍如何使用Hadoop MapReduce进行数据处理。什么是MapReduce？MapReduce是一种编程...

Hadoop MapReduce实战

前言

在大数据处理领域，Hadoop是一个非常重要的开源框架，它能够支持在廉价的硬件上运行大型分布式数据处理应用。Hadoop的核心组件之一是MapReduce，这是一种编程模型，用于大规模数据集（大于1TB）的并行处理。本文将通过一个具体的例子来介绍如何使用Hadoop MapReduce进行数据处理。

什么是MapReduce？

MapReduce是一种编程模型，用于处理和生成大数据集。用户指定一个map函数，用来处理键值对以生成一组中间键值对，以及一个reduce函数，用来合并所有具有相同中间键的中间值。MapReduce的设计理念是“分而治之”，它允许开发者轻松编写可扩展的应用程序，这些应用程序可以在数千台机器上并行运行。

环境准备

在开始我们的实战之前，需要确保已经安装了Hadoop环境。这里假设你已经有了一个运行良好的Hadoop集群。如果没有，可以参考官方文档进行安装配置。

实战案例：单词计数

案例背景

单词计数是一个经典的MapReduce示例，其目的是统计大量文本文件中每个单词出现的次数。这个简单的任务可以帮助我们理解MapReduce的基本工作原理。

编写MapReduce程序

Mapper阶段

Mapper的任务是从输入数据中提取出键值对。在这个例子中，键是文件中的每一行，值是这一行的内容。我们的目标是从每一行中提取出单词，并为每个单词生成一个键值对，其中键是单词本身，值是1（表示该单词出现了一次）。

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 {
            String line = value.toString();
            StringTokenizer itr = new StringTokenizer(line);
            while (itr.hasMoreTokens()) {
                word.set(itr.nextToken());
                context.write(word, one);
            }
        }
    }
}
Reducer阶段

Reducer的任务是处理来自Mapper的中间结果。在这个例子中，Reducer接收键值对，其中键是单词，值是一个整数列表，表示该单词出现的次数。Reducer的任务是将这些次数相加，得到最终的单词计数结果。

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);
    }
}

运行程序

编写完MapReduce程序后，需要将其打包成JAR文件，并提交到Hadoop集群上执行。可以通过以下命令完成：

hadoop jar your-jar-file.jar com.your.package.WordCount input-path output-path

这里的​​input-path​​是你想要处理的数据所在的目录，​​output-path​​是你希望存储结果的目录。

结果分析

程序执行完毕后，可以在​​output-path​​目录下找到结果文件。通常情况下，结果会以多个部分文件的形式存在，例如​​part-r-00000​​等。打开其中一个文件，可以看到类似如下的内容：

hello 3
world 2
big 1
data 1

这表示单词"hello"出现了3次，"world"出现了2次，等等。


Hadoop MapReduce 是一个用于处理和生成大数据集的编程模型。它通过将任务分解为多个小任务并行处理来提高效率。下面我将提供一个简单的 Hadoop MapReduce 实战示例，该示例用于计算文本文件中每个单词出现的次数。

示例场景

假设我们有一个文本文件 ​​input.txt​​，内容如下：

Hello World Hello Hadoop MapReduce Hello again world

我们的目标是使用 Hadoop MapReduce 来统计每个单词出现的次数。

步骤

  1. 编写 Mapper 类：Mapper 负责处理输入数据，并将其转换成键值对。
  2. 编写 Reducer 类：Reducer 负责汇总 Mapper 产生的中间结果。
  3. 编写 Driver 类：Driver 类配置作业参数并提交作业到 Hadoop 集群。

示例代码

1. Mapper 类
import java.io.IOException;
import java.util.StringTokenizer;

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

public class WordCountMapper extends Mapper<LongWritable, Text, Text, IntWritable> {
    private final static IntWritable one = new IntWritable(1);
    private Text word = new Text();

    @Override
    public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        String line = value.toString();
        StringTokenizer tokenizer = new StringTokenizer(line);
        while (tokenizer.hasMoreTokens()) {
            word.set(tokenizer.nextToken());
            context.write(word, one);
        }
    }
}
2. Reducer 类
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 WordCountReducer extends Reducer<Text, IntWritable, Text, IntWritable> {

    @Override
    public void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
        int sum = 0;
        for (IntWritable val : values) {
            sum += val.get();
        }
        context.write(key, new IntWritable(sum));
    }
}
3. 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 {
        if (args.length != 2) {
            System.err.println("Usage: WordCount <input path> <output path>");
            System.exit(-1);
        }

        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);
    }
}

运行步骤

  1. 编译代码：将上述代码保存到相应的文件中（如 WordCountMapper.java, WordCountReducer.java, WordCountDriver.java），然后编译这些 Java 文件。
javac -classpath `hadoop classpath` -d . WordCountMapper.java WordCountReducer.java WordCountDriver.java
  1. 打包：将编译后的类文件打包成 JAR 文件。
jar cf wordcount.jar *.class
  1. 运行 MapReduce 作业：使用 Hadoop 命令行工具运行作业。
hadoop jar wordcount.jar WordCountDriver /input/path /output/path
  1. 查看结果：作业完成后，可以在指定的输出路径查看结果文件。

这个示例展示了如何使用 Hadoop MapReduce 处理文本文件，统计每个单词出现的次数。你可以根据具体需求修改和扩展这个基本框架。当然可以！Hadoop MapReduce 是一个用于处理和生成大规模数据集的编程模型及其相关的实现。它通过将任务分解成多个小的子任务，并在多个计算节点上并行执行这些子任务来加速处理过程。MapReduce 框架由两个主要阶段组成：Map 阶段和 Reduce 阶段。

基本概念

  1. Map 阶段：在这个阶段，输入的数据被分割成多个块（通常称为 split），每个块会被分配给一个 map 任务处理。map 任务对输入的数据进行处理，输出一系列的键值对（key-value pairs）。
  2. Shuffle 和 Sort 阶段：这个阶段不是严格意义上的一个独立阶段，而是 Map 和 Reduce 阶段之间的过渡。在这个过程中，MapReduce 框架会收集所有 map 任务产生的中间结果，并按照 key 进行排序，然后将相同 key 的 value 分组，为 reduce 任务准备输入。
  3. Reduce 阶段：reduce 任务接收来自 shuffle 和 sort 阶段的输入，即一组带有相同 key 的 value 列表。reduce 任务对这些值进行汇总、聚合等操作，产生最终的输出结果。

示例代码

下面是一个简单的 WordCount 程序的例子，这是学习 MapReduce 经典的入门示例，用于统计文本文件中每个单词出现的次数。

1. Map 类
import java.io.IOException;
import java.util.StringTokenizer;

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

public class WordCountMapper extends Mapper<LongWritable, Text, Text, IntWritable> {

  private final static IntWritable one = new IntWritable(1);
  private Text word = new Text();

  public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
    String line = value.toString();
    StringTokenizer tokenizer = new StringTokenizer(line);
    while (tokenizer.hasMoreTokens()) {
      word.set(tokenizer.nextToken());
      context.write(word, one);
    }
  }
}
2. 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;

public class WordCountReducer extends Reducer<Text, IntWritable, Text, IntWritable> {

  public void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
    int sum = 0;
    for (IntWritable val : values) {
      sum += val.get();
    }
    context.write(key, new IntWritable(sum));
  }
}
3. 驱动程序
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 {
    if (args.length != 2) {
      System.err.println("Usage: WordCount <input path> <output path>");
      System.exit(-1);
    }

    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);
  }
}

代码解释

  • WordCountMapper：读取每行文本，使用 ​​StringTokenizer​​ 将行拆分为单词，对于每个单词，输出一个 ​​<word, 1>​​ 的键值对。
  • WordCountReducer：接受一组具有相同 key 的 value 列表，计算总和，并输出 ​​<word, total_count>​​。
  • WordCountDriver：设置作业配置，指定输入输出路径，以及使用的 mapper 和 reducer 类。

以上就是 Hadoop MapReduce 实战中 WordCount 示例的基本介绍和代码实现。希望这能帮助你理解如何编写和运行一个基本的 MapReduce 程序。如果你有更多具体的问题或需要进一步的帮助，请随时提问！  

【声明】本内容来自华为云开发者社区博主，不代表华为云及华为云开发者社区的观点和立场。转载时必须标注文章的来源（华为云社区）、文章链接、文章作者等基本信息，否则作者和本社区有权追究责任。如果您发现本社区中有涉嫌抄袭的内容，欢迎发送邮件进行举报，并提供相关证据，一经查实，本社区将立刻删除涉嫌侵权内容，举报邮箱： cloudbbs@huaweicloud.com
Hadoop MapReduce
  • 点赞
  • 收藏
  • 关注作者

评论（0）

0/1000
抱歉，系统识别当前为高风险访问，暂不支持该操作

全部回复

上滑加载中

设置昵称

在此一键设置昵称，即可参与社区互动！

*长度不超过10个汉字或20个英文字符，设置后3个月内不可修改。

*长度不超过10个汉字或20个英文字符，设置后3个月内不可修改。

确认 取消

加入云驻计划，成为创作者

  • 华为云周边好礼
  • 免费体验产品
  • 特殊身份标识
  • 线下官方门票
  • 内部专家零距离
  • 与10000+优质创作者共同成长
立即加入