hadoop基础之MapReduce的学习

hadoop基础之MapReduce的学习

MapReduce的执行步骤：

1.Map

package com.shujia.mr.worcount;

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;

/*
    TODO MapTask阶段
        自定义类继承Mapper,该Mapper类为一个具体的类，并其中定义了一些泛型
            
        MapTask阶段需要编写map函数，定义数据处理的逻辑
        KEYIN: 表示输入的Key的类型 表示map函数处理的Key类型 变量保存的数据是偏移量
                    读取数据的位置 字节数的位置非常大，需要使用Long类型 => LongWritable
        VALUEIN: 表示输入的Value类型  表示map函数处理的Value类型  表示的是一行字符串数据 String => Text
        KEYOUT: 表示输出的Key的类型 根据要处理的数据逻辑来进行定义 => 输出的Key为单词 => Java中的String类型 => Hadoop中的Text
        VALUEOUT:表示输出的Value的类型 根据要处理的数据逻辑来进行定义 => 输出的Value为1 => Java中的int类型 => Hadoop中的IntWritable

        注意:当数据在Hadoop中进行传递时，需要进行序列化，而Java中的序列化内容多，比较重，导致网络IO开销大
              为了计算速度快，Hadoop提供一套新的序列化类型
 */

// Mapper 中传入参数的类型由所要解决的问题来决定
// Mapper
public class WordCountMapper extends Mapper<LongWritable, Text,Text, IntWritable> {
    /**
     *  map函数中定义了Task任务在Map阶段所做的数据处理任务
     *      当前函数中需要对获取到的一行字符串进行按照 空格切分，再将单词遍历 之后再形成 Key为单词  1为Value的数据形式
     *  TODO 注意：map方法在执行的过程中是一行数据对应调用一次该函数
     * @param key 变量保存的数据是偏移量
     * @param value 表示的是一行字符串数据 是从文本文件中按行读取出来的
     * @param context 表示的是 Mapper.Context的上下文对象，作用是连接 Map阶段和Reduce阶段的桥梁
     */

    @Override
    protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, IntWritable>.Context context) throws IOException, InterruptedException {
        // value遍历中的数据 => hello hadoop
        // TODO 获取到的一行字符串进行按照 空格切分
        String[] words = value.toString().split(" ");

        // TODO 再将单词遍历
        for (String word : words) {
            // TODO 形成Key为单词  1为Value的数据形式
            // context 对象可以将Map阶段生成的数据发送给reduce阶段
            context.write(new Text(word),new IntWritable(1));
        }
    }
}

2.Reduce

package com.shujia.mr.worcount;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;

/*
    TODO ReduceTask阶段
        自定义类继承Reducer,该Reducer类为一个具体的类，并其中定义了一些泛型
            
        Reduce阶段的数据是由Map阶段发送过来的，所以Map阶段输出的类型就是Reduce阶段接收的类型
        根据处理逻辑：
            KEYIN: Text
            VALUEIN: IntWritable
        根据数据最终的要求：
            KEYOUT, VALUEOUT 表示最终每个单词出现的次数
            KEYOUT : Text
            VALUEOUT: IntWritable
 */
public class WordCountReducer extends Reducer<Text, IntWritable,Text, IntWritable> {

    /**
     *  reduce函数中定义了 Reduce阶段中要执行的代码逻辑
     *      将相同单词的KeyValue数据汇集到一起，再将所有的Value值 1 进行相加 得到最终的结果
     *  TODO 注意：① 对于reduce函数需要等Mapper阶段执行完成后才能再执行
     *            ② 对于每个Key会调用一次reduce函数
     *            ③ 对于Key的处理是存在有先后顺序的 按照字典序进行排序
     * @param key  表示map端输出的Key数据 单词
     * @param values 类型为Iterable 表示相同Key的Value数据形成的迭代器
     * @param context 上下文对象  可以将数据写出到HDFS
     * @throws IOException
     * @throws InterruptedException
     */

    @Override
    protected void reduce(Text key, Iterable<IntWritable> values, Reducer<Text, IntWritable, Text, IntWritable>.Context context) throws IOException, InterruptedException {
        // 定义num 用于记录单词出现的次数
        int num = 0;
        // TODO 再将所有的Value值 1 进行相加 得到最终的结果
        for (IntWritable value : values) {
            // value为IntWritable类型，需要使用get()取出其中的数值，再进行相加
            num += value.get();
        }
        context.write(key,new IntWritable(num));

    }
}

3.MapReduce程序入口中的固定写法

package com.shujia.mr.worcount;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
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.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;

import java.io.FileNotFoundException;
import java.io.IOException;

public class WordCount {
    public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {
        // TODO MapReduce程序入口中的固定写法

        // TODO 1.获取Job对象 并设置相关Job任务的名称及入口类
        // 方式1：
//        Job job = new Job();
//        job.setJobName("word count");
        // 方式2：（常用）
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf, "word count");
        // 设置当前main方法所在的入口类
        job.setJarByClass(WordCount.class);

        // TODO 2.设置自定义的Mapper和Reducer类
        job.setMapperClass(WordCountMapper.class);
        job.setReducerClass(WordCountReducer.class);

        // TODO 3.设置Mapper的KeyValue输出类 和 Reducer的输出类 (最终输出)
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(IntWritable.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);

        //TODO 4.设置数据的输入和输出路径
        //  输入、输出都在HDFS上？？？（执行方式1）
        //  org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
        //  org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
        // 数据的输入：读取HDFS中的数据
//        TextInputFormat.addInputPath(job,new Path("/data/words.txt"));
//        TextOutputFormat.setOutputPath(job,new Path("/api/wordCount"));

        // 本地路径（执行方式2：在本地执行）
        // 从job中获取Configuration对象，获取一个fileSystem对象
        FileSystem fileSystem = FileSystem.get(job.getConfiguration());
        // 定义路径（并且下面会对路径的存在与否进行判断）
        Path outPath = new Path("hadoop/out/wordCount");
//        Path inpath = new Path("hadoop/data/words.txt");
        Path inpath = new Path("hadoop/data/words");

        // 判断输入路径是否存在
        if (!fileSystem.exists(inpath)) {
            throw new FileNotFoundException(inpath+"不存在");
//            System.out.println(inpath+"不存在");
//            System.exit(1);
        }


        // TODO TextInputFormat、FileInputFormat类均可实现该方法
        //  添加输入路径
//        TextInputFormat.addInputPath(job,inpath);
        FileInputFormat.addInputPath(job,inpath);

        // 判断输出路径是否存在，若存在则进行删除
        if (fileSystem.exists(outPath)) {
            System.out.println("路径存在，开始删除");
            fileSystem.delete(outPath,true);
        }


        // TODO TextInputFormat、FileInputFormat类均可实现该方法
        //  添加输出路径
//        TextOutputFormat.setOutputPath(job,outPath);
        FileOutputFormat.setOutputPath(job,outPath);


        // TODO 5.提交任务开始执行
        job.waitForCompletion(true);
    }
}