Hadoop框架：MapReduce基本原理和入门案例 _MapReduce

一、MapReduce概述1、基本概念Hadoop核心组件之一：分布式计算的方案MapReduce ，是一种编程模型，用于大规模数据集的并行运算，其中Map（映射）和Reduce（归约）。
MapReduce既是一个编程模型，也是一个计算组件，处理的过程分为两个阶段， Map阶段：负责把任务分解为多个小任务， Reduce负责把多个小任务的处理结果进行汇总。其中Map阶段主要输入是一对Key-Value ，经过map计算后输出一对Key-Value值；然后将相同Key合并，形成Key-Value集合；再将这个Key-Value集合转入Reduce阶段，经过计算输出最终Key-Value结果集。
2、特点描述MapReduce可以实现基于上千台服务器并发工作，提供很强大的数据处理能力，如果其中单台服务挂掉，计算任务会自动转义到另外节点执行，保证高容错性；但是MapReduce不适应于实时计算与流式计算，计算的数据是静态的。
二、操作案例1、流程描述

文章插图

数据文件一般以CSV格式居多，数据行通常以空格分隔，这里需要考虑数据内容特点；
文件经过切片分配在不同的MapTask任务中并发执行；
MapTask任务执行完毕之后，执行ReduceTask任务，依赖Map阶段的数据；
ReduceTask任务执行完毕后，输出文件结果。
2、基础配置

hadoop:# 读取的文件源inputPath: hdfs://hop01:9000/hopdir/JAVANew.txt# 该路径必须是程序运行前不存在的outputPath: /wordOut

3、MApper程序

public class WordMapper extends Mapper<LongWritable, Text, Text, IntWritable> {Text mapKey = new Text();IntWritable mapValue = https://www.isolves.com/it/cxkf/kj/2020-11-23/new IntWritable(1);@Overrideprotected void map (LongWritable key, Text value, Context context)throws IOException, InterruptedException {// 1、读取行String line = value.toString();// 2、行内容切割 ， 根据文件中分隔符String[] words = line.split(" ");// 3、存储for (String word : words) {mapKey.set(word);context.write(mapKey, mapValue);}}}

4、Reducer程序

public class WordReducer extends Reducer<Text, IntWritable, Text, IntWritable> {int sum ;IntWritable value = https://www.isolves.com/it/cxkf/kj/2020-11-23/new IntWritable();@Overrideprotected void reduce(Text key, Iterable values,Context context)throws IOException, InterruptedException {// 1、累加求和统计sum = 0;for (IntWritable count : values) {sum += count.get();}// 2、输出结果value.set(sum);context.write(key,value);}}

5、执行程序

@RestControllerpublic class WordWeb {@Resourceprivate MapReduceConfig mapReduceConfig ;@GetMapping("/getWord")public String getWord () throws IOException, ClassNotFoundException, InterruptedException {// 声明配置Configuration hadoopConfig = new Configuration();hadoopConfig.set("fs.hdfs.impl",org.Apache.hadoop.hdfs.DistributedFileSystem.class.getName());hadoopConfig.set("fs.file.impl",org.apache.hadoop.fs.LocalFileSystem.class.getName());Job job = Job.getInstance(hadoopConfig);// Job执行作业 输入路径FileInputFormat.addInputPath(job, new Path(mapReduceConfig.getInputPath()));// Job执行作业 输出路径FileOutputFormat.setOutputPath(job, new Path(mapReduceConfig.getOutputPath()));// 自定义 Mapper和Reducer 两个阶段的任务处理类job.setMapperClass(WordMapper.class);job.setReducerClass(WordReducer.class);// 设置输出结果的Key和Value的类型job.setOutputKeyClass(Text.class);job.setOutputValueClass(IntWritable.class);//执行Job直到完成job.waitForCompletion(true);return "success" ;}}

6、执行结果查看将应用程序打包放到hop01服务上执行；
java -jar map-reduce-case01.jar

文章插图

三、案例分析1、数据类型Java数据类型与对应的Hadoop数据序列化类型；

文章插图

2、核心模块Mapper模块：处理输入的数据，业务逻辑在map()方法中完成，输出的数据也是KV格式；
Reducer模块：处理Map程序输出的KV数据，业务逻辑在reduce()方法中；
Driver模块：将程序提交到yarn进行调度，提交封装了运行参数的job对象；
四、序列化操作1、序列化简介序列化：将内存中对象转换为二进制的字节序列，可以通过输出流持久化存储或者网络传输；
反序列化：接收输入字节流或者读取磁盘持久化的数据，加载到内存的对象过程；