Hadoop系列Hadoop三大核心之MapReduce-程序编写

Posted 南山饱虎

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了Hadoop系列Hadoop三大核心之MapReduce-程序编写相关的知识,希望对你有一定的参考价值。

接下来以一个简单的WordCount为例子,介绍Java版本的MapReduce的程序编写。

mapreduce程序主要分三部分:1.map部分,2.reduce部分,3.提交部分。

1. 准备部分

hadoop中,针对数据类型自成一体,与java的数据类型对应。封装在hadoop.io包中,主要分为基本类型和其它类型。

  • 基本数据类型
数据类型 hadoop数据类型 Java数据类型
布尔 BooleanWritable boolean
整型 IntWritable int
长整型 LongWritable long
浮点型 FloatWritable float
双精浮点 DoubleWritable double
字节 ByteWritable byte
  • 其它类型
数据类型 hadoop数据类型 Java数据类型
字符串 Text String
数组 ArrayWritable Array
Map MapWritable Map

2. jar包依赖

创建一个maven工程,pom.xml文件中,添加以下依赖

<dependencies>
    <dependency>
        <groupId>org.apache.hadoop</groupId>
        <artifactId>hadoop-common</artifactId>
        <version>2.6.0</version>
    </dependency>

    <dependency>
        <groupId>org.apache.hadoop</groupId>
        <artifactId>hadoop-hdfs</artifactId>
        <version>2.6.0</version>
    </dependency>

    <dependency>
        <groupId>org.apache.hadoop</groupId>
        <artifactId>hadoop-client</artifactId>
        <version>2.6.0</version>
    </dependency>
</dependencies>

3. Map部分

映射部分,将数据逐条处理

首先,需要继承Mapper<KEYIN, VALUEIN, KEYOUT, VALUEOUT>类

四个泛型参数分别代表:输入key 输出value 输出key 输出value

然后重写mapper的map方法

map(KEYIN key, VALUEIN value, Context context) throws IOException, InterruptedException

主体代码:

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

    /**
     *  流程,输入key和value,map的结果写入到context中
     */
    @Override
    public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        //读取一行数据
        String line = value.toString();
        //因为英文字母是以“ ”为间隔的,因此使用“ ”分隔符将一行数据切成多个单词并存在数组中
        String str[] = line.split(" ");
        //循环迭代字符串,将一个单词变成<key,value>形式,及<"hello",1>
        for (String s : str) {
            context.write(new Text(s), new IntWritable(1));
        }
    }

}

4.Reduce部分

归并部分,将map处理和shuffle之后的数据进行归并。shuffle过程由hadoop控制

Reducer<KEYIN,VALUEIN,KEYOUT,VALUEOUT>

四个泛型参数分别代表:输入key 输出value 输出key 输出value

然后重写reducer的reduce方法

reduce(KEYIN key, Iterable<VALUEIN> values, Context context) throws IOException, InterruptedException

主体代码:

public class WordCountReduce extends Reducer<Text,IntWritable,Text, IntWritable> {
    /**
     *  流程,输入key和value,map的结果写入到context中
     */
    @Override
    public void reduce(Text key, Iterable<IntWritable> values,Context context)throws IOException,InterruptedException{
        int count = 0;

        for(IntWritable value: values) {
            count++;
        }
        context.write(key,new IntWritable(count));
    }
}

5.提交部分

mapreduce的入口

public static void main(String[] args) throws Exception{
    Configuration conf = new Configuration();
    // 构造一个job对象来封装本mapreduce业务到所有信息
    Job mrJob = Job.getInstance(conf);
    // 指定本job工作用到的jar包位置
    mrJob.setJarByClass(WordCount.class);
    // 指定本job用到的mapper类
    mrJob.setMapperClass(WordCountMap.class);
    // 指定本job用到的reducer类
    mrJob.setReducerClass(WordCountReduce.class);

    // 指定mapper输出的kv类型
    mrJob.setMapOutputKeyClass(Text.class);
    mrJob.setMapOutputValueClass(IntWritable.class);


    // 指定reducer输出到kv数据类型(setOutputKeyClass 会对mapper和reducer都起作用,如果上面mapper不设置的话)
    mrJob.setOutputKeyClass(Text.class);
    mrJob.setOutputValueClass(IntWritable.class);


    FileInputFormat.addInputPath(mrJob,new Path(args[0]));

    //设置mapreduce程序的输出路径,MapReduce的结果都是输入到文件中
    FileOutputFormat.setOutputPath(mrJob,new Path(args[1]));

    // 最后提交任务
    boolean waitForCompletion = mrJob.waitForCompletion(true);
    System.exit(waitForCompletion ? 0 : 1);
}

6.打包提交

将程序打成jar包,提交到hadoop集群中,然后使用命令行进行任务提交

对于输入输出路径,均可接受本地路径和hdfs路径。

本地路径前缀:file://

hdfs路径前缀:hdfs://

/hadoop-3.1.2/bin/hadoop jar my_mapreduce-1.0-SNAPSHOT.jar com.breakthrough.wordcount.WordCount file:///home/hadoop/english.txt file:///home/hadoop/output

以上是关于Hadoop系列Hadoop三大核心之MapReduce-程序编写的主要内容,如果未能解决你的问题,请参考以下文章

Hadoop系列Hadoop三大核心之MapReduce-程序编写

Hadoop的三大核心组件之HDFS和YARN

Hadoop之MapReduce命令

Hadoop之 - 剖析 MapReduce 作业的运行机制(MapReduce 2)

Hadoop之YARN的安装与测试

Hadoop集群配置