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类为一个具体的类，并其中定义了一些泛型<KEYIN, VALUEIN, KEYOUT, VALUEOUT>MapTask阶段需要编写map函数，定义数据处理的逻辑KEYIN: 表示输入的Key的类型 表示map函数处理的Key类型 变量保存的数据是偏移量读取数据的位置 字节数的位置非常大，需要使用Long类型 => LongWritableVALUEIN: 表示输入的Value类型 表示map函数处理的Value类型 表示的是一行字符串数据 String => TextKEYOUT: 表示输出的Key的类型 根据要处理的数据逻辑来进行定义 => 输出的Key为单词 => Java中的String类型 => Hadoop中的TextVALUEOUT:表示输出的Value的类型 根据要处理的数据逻辑来进行定义 => 输出的Value为1 => Java中的int类型 => Hadoop中的IntWritable注意:当数据在Hadoop中进行传递时，需要进行序列化，而Java中的序列化内容多，比较重，导致网络IO开销大为了计算速度快，Hadoop提供一套新的序列化类型*/// Mapper<KEYIN, VALUEIN, KEYOUT, VALUEOUT> 中传入参数的类型由所要解决的问题来决定
// Mapper<LongWritable, Text, Text, IntWritable>
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阶段的桥梁*/@Overrideprotected 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类为一个具体的类，并其中定义了一些泛型<KEYIN, VALUEIN, KEYOUT, VALUEOUT>Reduce阶段的数据是由Map阶段发送过来的，所以Map阶段输出的类型就是Reduce阶段接收的类型根据处理逻辑：KEYIN: TextVALUEIN: IntWritable根据数据最终的要求：KEYOUT, VALUEOUT 表示最终每个单词出现的次数KEYOUT : TextVALUEOUT: 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*/@Overrideprotected 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);}
}