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