3.1 Hadoop-MapReduce编程逻辑

逃离我推掉我的手 2022-05-24 00:39 124阅读 0赞

Hadoop MapReduce工作详细流程(Partitioner/SortComparator/GroupingComparator）:

*  map阶段

1. 使用job.setInputFormatClass(TextInputFormat)做为输入格式。注意输出应该符合自定义Map中定义的输出。

2. 进入Mapper的map()方法，生成一个List。

3. 在map阶段的最后，会先调用 job.setPartitionerClass()对这个List进行分区，每个分区映射到一个reducer。

4. 每个分区内又调用 job.setSortComparatorClass()设置的key比较函数类排序(如果没有通过job.setSortComparatorClass()设置key比较函数类，则使用key的实现的compareTo方法)。可以看到，这是一个二次排序。

5. 如果设置了Combiner（job.setCombinerClass）对output进行一次合并，从而减少对reduce的输出流量和预处理reduce的input数据。但不一定会执行，对于Combiner执行时机参考Reference\[4\]。

【说明】以上步骤省略了collect阶段、cache阶段等细节，更详细步骤参考Reference\[3\]

*  reduce阶段

1. shuffle阶段

reducer开始fetch所有映射到这个reducer的map输出。

2.1 sort阶段

再次调用 job.setSortComparatorClass()设置的key比较函数类对所有数据对排序(因为一个reducer接受多个mappers，需要重新排序)。

2.2 secondary sort阶段

然后开始构造一个key对应的value迭代器。这时就要用到分组，使用 jobjob.setGroupingComparatorClass()设置的分组函数类。只要这个比较器比较的两个key相同，他们就属于同一个组，它们的value放在一个value迭代器，而这个迭代器的key使用属于同一个组的所有key的第一个key。

3.reduce阶段

最后就是进入Reducer的reduce()方法，reduce()方法的输入是所有的（key和它的value迭代器）。同样注意输入与输出的类型必须与自定义的Reducer中声明的一致。

【注意】reducers的输出是无序的。

--------------------

System.setProperty("HADOOP\_USER\_NAME", "root");//设置操作Hadoop用户

1、MR执行环境有两种：本地测试环境，服务器环境

本地测试环境(windows)：

在windows的hadoop目录bin目录有一个winutils.exe

1、在windows下配置hadoop的环境变量

2、拷贝debug工具(winutils.exe)到HADOOP\_HOME/bin

3、修改hadoop的源码 ，注意：确保项目的lib需要真实安装的jdk的lib

4、MR调用的代码需要改变：

a、src不能有服务器的hadoop配置文件

b、在调用是使用：

Configuration config = new Configuration();

config.set("fs.defaultFS", "hdfs://node7:8020");

注意：CDH搭建的端口一般默认8082，普通搭建9000

服务器环境：

首先需要在src下放置服务器上的hadoop配置文件

1、在本地直接调用，执行过程在服务器上（真正企业运行环境）

a、把MR程序打包（jar），直接放到本地

b、修改hadoop的源码 ，注意：确保项目的lib需要真实安装的jdk的lib

c、增加一个属性：

config.set("mapred.jar", "C:\\\\Users\\\\Administrator\\\\Desktop\\\\wc.jar");

d、本地执行main方法，servlet调用MR。

2、直接在服务器上，使用命令的方式调用，执行过程也在服务器上

a、把MR程序打包（jar），传送到服务器上

b、通过： hadoop jar jar路径 类的全限定名

hadoop jar MapReduce.jar com.test.mapreduce.RunJob

本地配置eclipse运行MR：

1，配置Hadoop环境

![20180521152505891][]

2，添加相对应的插件

![20180521152515105][]

3，添加相应的源码

**Exception in thread "main"java.lang.UnsatisfiedLinkError:org.apache.hadoop.io.nativeio.NativeIO$Windows.access0(Ljava/lang/String;I)Z错误：**

![20180521152525500][]

![20180521152533933][]

[20180521152505891]: /images/20220524/9afddc5f1b9c4b9cad678956b1c4f850.png
[20180521152515105]: /images/20220524/3bcb69a8f80d404bbcdb7c3793db9487.png
[20180521152525500]: /images/20220524/cfe7412653874e61b2fba7ddfc0194b6.png
[20180521152533933]: /images/20220524/ded2d405467f4d89b8519f9d6bb434a9.png