MapReduce的运行流程概述

2020-07-14 19:34:56 浏览数 (1)

MapReduce处理数据的大致流程

InputFormat调用RecordReader,从输入目录的文件中,读取一组数据,封装为keyin-valuein对象

②将封装好的key-value,交给Mapper.map()------>将处理的结果写出 keyout-valueout

ReduceTask启动Reducer,使用Reducer.reduce()处理Mapper的keyout-valueout

④OutPutFormat调用RecordWriter,将Reducer处理后的keyout-valueout写出到文件

示例

需求: 统计/hello目录中每个文件的单词数量, a-p开头的单词放入到一个结果文件中, q-z开头的单词放入到另外一个结果文件中。

例如: /hello/a.txt,文件大小200M hello,hi,hadoop hive,hadoop,hive, zoo,spark,wow zoo,spark,wow ...

/hello/b.txt,文件大小100M hello,hi,hadoop zoo,spark,wow ...

1. Map阶段(运行MapTask,将一个大的任务切分为若干小任务,处理输出阶段性的结果)

①切片(切分数据) /hello/a.txt 200M /hello/b.txt 100M

默认的切分策略是以文件为单位,以文件的块大小(128M)为片大小进行切片! split0:/hello/a.txt,0-128M split1: /hello/a.txt,128M-200M split2: /hello/b.txt,0M-100M

②运行MapTask(进程),每个MapTask负责一片数据

split0:/hello/a.txt,0-128M--------MapTask1 split1: /hello/a.txt,128M-200M--------MapTask2 split2: /hello/b.txt,0M-100M--------MapTask3

③读取数据阶段

在MR中,所有的数据必须封装为key-value MapTask1,2,3都会初始化一个InputFormat(默认TextInputFormat),每个InputFormat对象负责创建一个RecordReader(LineRecordReader)对象, RecordReader负责从每个切片的数据中读取数据,封装为key-value

LineRecordReader: 将文件中的每一行封装为一个key(offset)-value(当前行的内容) 举例: hello,hi,hadoop----->(0,hello,hi,hadoop) hive,hadoop,hive----->(20,hive,hadoop,hive) zoo,spark,wow----->(30,zoo,spark,wow) zoo,spark,wow----->(40,zoo,spark,wow)

④进入Mapper的map()阶段

map()是Map阶段的核心处理逻辑! 单词统计! map()会循环调用,对输入的每个Key-value都进行处理! 输入:(0,hello,hi,hadoop) 输出:(hello,1),(hi,1),(hadoop,1)

输入:(20,hive,hadoop,hive) 输出:(hive,1),(hadoop,1),(hive,1)

输入:(30,zoo,spark,wow) 输出:(zoo,1),(spark,1),(wow,1)

输入:(40,zoo,spark,wow) 输出:(zoo,1),(spark,1),(wow,1)

⑤将MapTask输出的记录进行分区(分组、分类)

在Mapper输出后,调用Partitioner,对Mapper输出的key-value进行分区,分区后也会排序(默认字典顺序排序) 分区规则:

  • a-p开头的单词放入到一个区
  • q-z开头的单词放入到另一个区 MapTask1: 0号区: (hadoop,1),(hadoop,1),(hello,1),(hi,1),(hive,1),(hive,1) 1号区: (spark,1),(spark,1),(wow,1) ,(wow,1),(zoo,1)(zoo,1)

MapTask2: 0号区: ... 1号区: ...

MapTask3: 0号区: (hadoop,1),(hello,1),(hi,1), 1号区: (spark,1),(wow,1),(zoo,1)

2.Reduce阶段

①因为需求是生成两个结果文件,所以我们需要启动两个ReduceTask ReduceTask启动后,会启动shuffle线程,从MapTask中拷贝相应分区的数据!

ReduceTask1: 只负责0号区 将三个MapTask,生成的0号区数据全部拷贝到ReduceTask所在的机器! (hadoop,1),(hadoop,1),(hello,1),(hi,1),(hive,1),(hive,1) (hadoop,1),(hello,1),(hi,1),

ReduceTask2: 只负责1号区 将三个MapTask,生成的1号区数据全部拷贝到ReduceTask所在的机器! (spark,1),(spark,1),(wow,1) ,(wow,1),(zoo,1)(zoo,1) (spark,1),(wow,1),(zoo,1)

②sort

ReduceTask1: 只负责0号区进行排序: (hadoop,1),(hadoop,1),(hadoop,1),(hello,1),(hello,1),(hi,1),(hi,1),(hive,1),(hive,1) ReduceTask2: 只负责1号区进行排序: (spark,1),(spark,1),(spark,1),(wow,1) ,(wow,1),(wow,1),(zoo,1),(zoo,1)(zoo,1)

③reduce ReduceTask1---->Reducer----->reduce(一次读入一组数据)

何为一组数据: key相同的为一组数据 输入: (hadoop,1),(hadoop,1),(hadoop,1) 输出: (hadoop,3)

输入: (hello,1),(hello,1) 输出: (hello,2)

输入: (hi,1),(hi,1) 输出: (hi,2)

输入:(hive,1),(hive,1) 输出: (hive,2)

ReduceTask2---->Reducer----->reduce(一次读入一组数据)

输入: (spark,1),(spark,1),(spark,1) 输出: (spark,3)

输入: (wow,1) ,(wow,1),(wow,1) 输出: (wow,3)

输入:(zoo,1),(zoo,1)(zoo,1) 输出: (zoo,3)

④调用OutPutFormat中的RecordWriter将Reducer输出的记录写出 ReduceTask1---->OutPutFormat(默认TextOutPutFormat)---->RecordWriter(LineRecoreWriter) LineRecoreWriter将一个key-value以一行写出,key和alue之间使用t分割 在输出目录中,生成文件part-r-0000 hadoop 3 hello 2 hi 2 hive 2

ReduceTask2---->OutPutFormat(默认TextOutPutFormat)------>RecordWriter(LineRecoreWriter) LineRecoreWriter将一个key-value以一行写出,key和alue之间使用t分割 在输出目录中,生成文件part-r-0001 spark 3 wow 3 zoo 3

三、MR总结

Map阶段(MapTask): 切片(Split)-----读取数据(Read)-------交给Mapper处理(Map)------分区和排序(sort)

Reduce阶段(ReduceTask): 拷贝数据(copy)------排序(sort)-----合并(reduce)-----写出(write)

0 人点赞