本文源碼：GitHub·點這里 || GitEE·點這里

一、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 = 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 = new IntWritable();@Overrideprotected void reduce(Text key, Iterable<IntWritable> 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、執行程序

@RestController public 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數據序列化類型；

Java類型Writable類型Java類型Writable類型

String	Text	float	FloatWritable
int	IntWritable	long	LongWritable
boolean	BooleanWritable	double	DoubleWritable
byte	ByteWritable	array	DoubleWritable
map	MapWritable

2、核心模塊

Mapper模塊：處理輸入的數據，業務邏輯在map()方法中完成，輸出的數據也是KV格式；

Reducer模塊：處理Map程序輸出的KV數據，業務邏輯在reduce()方法中；

Driver模塊：將程序提交到yarn進行調度，提交封裝了運行參數的job對象；

四、序列化操作

1、序列化簡介

序列化：將內存中對象轉換為二進制的字節序列，可以通過輸出流持久化存儲或者網絡傳輸；

反序列化：接收輸入字節流或者讀取磁盤持久化的數據，加載到內存的對象過程；

Hadoop序列化相關接口：Writable實現的序列化機制、Comparable管理Key的排序問題；

2、案例實現

案例描述：讀取文件，并對文件相同的行做數據累加計算，輸出計算結果；該案例演示在本地執行，不把Jar包上傳的hadoop服務器，驅動配置一致。

實體對象屬性

public class AddEntity implements Writable {private long addNum01;private long addNum02;private long resNum;// 構造方法public AddEntity() {super();}public AddEntity(long addNum01, long addNum02) {super();this.addNum01 = addNum01;this.addNum02 = addNum02;this.resNum = addNum01 + addNum02;}// 序列化@Overridepublic void write(DataOutput dataOutput) throws IOException {dataOutput.writeLong(addNum01);dataOutput.writeLong(addNum02);dataOutput.writeLong(resNum);}// 反序列化@Overridepublic void readFields(DataInput dataInput) throws IOException {// 注意：反序列化順序和寫序列化順序一致this.addNum01 = dataInput.readLong();this.addNum02 = dataInput.readLong();this.resNum = dataInput.readLong();}// 省略Get和Set方法 }

Mapper機制

public class AddMapper extends Mapper<LongWritable, Text, Text, AddEntity> {Text myKey = new Text();@Overrideprotected void map(LongWritable key, Text value, Context context)throws IOException, InterruptedException {// 讀取行String line = value.toString();// 行內容切割String[] lineArr = line.split(",");// 內容格式處理String lineNum = lineArr[0];long addNum01 = Long.parseLong(lineArr[1]);long addNum02 = Long.parseLong(lineArr[2]);myKey.set(lineNum);AddEntity myValue = new AddEntity(addNum01,addNum02);// 輸出context.write(myKey, myValue);} }

Reducer機制

public class AddReducer extends Reducer<Text, AddEntity, Text, AddEntity> {@Overrideprotected void reduce(Text key, Iterable<AddEntity> values, Context context)throws IOException, InterruptedException {long addNum01Sum = 0;long addNum02Sum = 0;// 處理Key相同for (AddEntity addEntity : values) {addNum01Sum += addEntity.getAddNum01();addNum02Sum += addEntity.getAddNum02();}// 最終輸出AddEntity addRes = new AddEntity(addNum01Sum, addNum02Sum);context.write(key, addRes);} }

案例最終結果：

五、源代碼地址

GitHub·地址 https://github.com/cicadasmile/big-data-parent GitEE·地址 https://gitee.com/cicadasmile/big-data-parent

推薦閱讀：編程體系整理

序號項目名稱GitHub地址GitEE地址推薦指數

01	Java描述設計模式,算法,數據結構	GitHub·點這里	GitEE·點這里	☆☆☆☆☆
02	Java基礎、并發、面向對象、Web開發	GitHub·點這里	GitEE·點這里	☆☆☆☆
03	SpringCloud微服務基礎組件案例詳解	GitHub·點這里	GitEE·點這里	☆☆☆
04	SpringCloud微服務架構實戰綜合案例	GitHub·點這里	GitEE·點這里	☆☆☆☆☆
05	SpringBoot框架基礎應用入門到進階	GitHub·點這里	GitEE·點這里	☆☆☆☆
06	SpringBoot框架整合開發常用中間件	GitHub·點這里	GitEE·點這里	☆☆☆☆☆
07	數據管理、分布式、架構設計基礎案例	GitHub·點這里	GitEE·點這里	☆☆☆☆☆
08	大數據系列、存儲、組件、計算等框架	GitHub·點這里	GitEE·點這里	☆☆☆☆☆

總結

以上是生活随笔為你收集整理的Hadoop框架：MapReduce基本原理和入门案例的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。