Hadoop MapReduce & YARN

MapReduce

• 将要执行的需求，分解为多个Map Task和Reduce Task

• 将Map Task 和 Reduce Task分配到对应的服务器去执行

MapReduce模型

1. Map阶段：
• 输入数据被分成若干个分片，每个分片由一个Map任务处理。
• Map任务读取输入数据，并将其转换成键值对（key-value pairs）。
• 对每个输入键值对，执行用户定义的Map函数，产生中间键值对（intermediate key-value pairs）。

2. Shuffle and Sort阶段（隐含阶段）：
• Map任务产生的中间键值对被按照键进行排序和分组，具有相同键的所有键值对被聚合在一起。
• 这些中间键值对被传输到相应的Reduce任务。

3. Reduce阶段：
• Reduce任务处理来自一个或多个Map任务的中间键值对。
• 对每个中间键，执行用户定义的Reduce函数，将所有对应的中间值进行处理，产生最终的输出结果。

MapReduce的工作流程

1. 输入数据分割：
• 输入数据被分割成固定大小的分片（通常为HDFS块大小，默认128MB或256MB）。

2. Map任务执行：
• 每个Map任务处理一个数据分片，执行用户定义的Map函数，将输入数据转换成中间键值对。
• 中间键值对被写入本地磁盘，并进行分区和排序。

3. Shuffle and Sort：
• 中间键值对被按键进行排序，并通过网络传输到对应的Reduce任务。
• 具有相同键的所有中间键值对被聚合在一起。

4. Reduce任务执行：
• Reduce任务接收和处理来自多个Map任务的中间键值对，执行用户定义的Reduce函数，将中间值进行合并和处理，产生最终的输出结果。

5. 输出结果写入：
• 最终的输出结果被写入HDFS或其他分布式文件系统。

MapReduce部署配置

• 修改文件：vim /export/server/hadoop/etc/hadoop/mapred-env.sh，追加内容

# 设置 JDK 文件路径
export JAVA_HOME=/export/server/jdk
# 设置历史服务器进程内存
export HADOOP_JOB_HISTORYSERVER_HEAPSIZE=1000
# 设置日志级别为 INFO
export HADOOP_ROOT_LOGGER=INFO,RFA

• 修改文件：vim /export/server/hadoop/etc/hadoop/mapred-site.xml，修改内容

<configuration>
  <!-- 
  配置项名称：mapreduce.framework.name
  含义：指定使用的MapReduce框架名称，这里使用YARN作为资源管理器。
  参数：yarn（使用YARN作为框架）
  -->
  <property>
    <name>mapreduce.framework.name</name>
    <value>yarn</value>
  </property>

  <!-- 
  配置项名称：mapreduce.jobhistory.address
  含义：指定JobHistory服务器的地址，用于跟踪MapReduce作业的历史信息。
  参数：anjhon:10020（JobHistory服务器的主机和端口）
  -->
  <property>
    <name>mapreduce.jobhistory.address</name>
    <value>anjhon:10020</value>
  </property>

  <!-- 
  配置项名称：mapreduce.jobhistory.webapp.address
  含义：指定JobHistory Web界面的地址，用于查看作业的历史信息。
  参数：anjhon:19888（JobHistory Web界面的主机和端口）
  -->
  <property>
    <name>mapreduce.jobhistory.webapp.address</name>
    <value>anjhon:19888</value>
  </property>

  <!-- 
  配置项名称：mapreduce.jobhistory.intermediate-done-dir
  含义：指定中间完成作业的目录，用于存储作业执行的临时数据。
  参数：/data/mr-history/tmp（中间完成目录路径）
  -->
  <property>
    <name>mapreduce.jobhistory.intermediate-done-dir</name>
    <value>/data/mr-history/tmp</value>
  </property>

  <!-- 
  配置项名称：mapreduce.jobhistory.done-dir
  含义：指定作业完成后存储历史记录的目录。
  参数：/data/mr-history/done（完成记录目录路径）
  -->
  <property>
    <name>mapreduce.jobhistory.done-dir</name>
    <value>/data/mr-history/done</value>
  </property>

  <!-- 
  配置项名称：yarn.app.mapreduce.am.env
  含义：指定MapReduce应用程序主控进程（AM）的环境变量。
  参数：HADOOP_MAPRED_HOME=$HADOOP_HOME（设置HADOOP_MAPRED_HOME变量为HADOOP_HOME）
  -->
  <property>
    <name>yarn.app.mapreduce.am.env</name>
    <value>HADOOP_MAPRED_HOME=$HADOOP_HOME</value>
  </property>

  <!-- 
  配置项名称：mapreduce.map.env
  含义：指定Map任务的环境变量。
  参数：HADOOP_MAPRED_HOME=$HADOOP_HOME（设置HADOOP_MAPRED_HOME变量为HADOOP_HOME）
  -->
  <property>
    <name>mapreduce.map.env</name>
    <value>HADOOP_MAPRED_HOME=$HADOOP_HOME</value>
  </property>

  <!-- 
  配置项名称：mapreduce.reduce.env
  含义：指定Reduce任务的环境变量。
  参数：HADOOP_MAPRED_HOME=$HADOOP_HOME（设置HADOOP_MAPRED_HOME变量为HADOOP_HOME）
  -->
  <property>
    <name>mapreduce.reduce.env</name>
    <value>HADOOP_MAPRED_HOME=$HADOOP_HOME</value>
  </property>
</configuration>

MapReduce的特点和优势

1. 简化编程模型：
• MapReduce将复杂的分布式计算过程抽象为简单的Map和Reduce函数，用户只需关注具体的业务逻辑。

2. 高可扩展性：
• MapReduce能够处理大规模数据集，通过增加计算节点可以水平扩展，处理TB级、PB级的数据。

3. 容错性：
• MapReduce框架具有内建的容错机制，自动处理任务失败和数据丢失，通过重试和数据冗余确保计算的可靠性。

4. 数据本地化：
• MapReduce框架尽量将计算任务分配到存储数据的节点上，减少数据传输，提高计算效率。

YARN

YARN核心组件

ResourceManager

• YARN集群的核心组件，负责整个集群的资源管理和调度。

• 包括两个重要子组件：
• Scheduler：负责资源的分配，不直接监控和追踪任务，只根据资源的使用策略（如容量调度器、公平调度器）来分配资源。
• ApplicationManager：负责管理应用程序的提交和生命周期，包括为应用程序启动第一个容器（ApplicationMaster）。
• 每个应用程序（如MapReduce作业）的实例化管理进程，负责应用程序的具体任务调度和协调。
• 与ResourceManager交互申请资源，并与NodeManager交互以启动和监控任务容器。

NodeManager

• 每个节点上的资源管理守护进程，负责节点上的资源管理、容器的启动和监控。

• 向ResourceManager汇报节点的资源使用情况和容器状态。

• Container：NodeManager为不同的应用程序创建容器，以供其运行
• Container 是 YARN 中的资源抽象，它封装了某个节点上的多维度资源，如内存、CPU、磁盘、网络等。
• 由NodeManager管理，每个Container运行一个特定的任务。

YARN辅助组件

代理服务器：Web Application Proxy

历史服务器：JobHistoryServer

YARN的工作流程

1. 应用程序提交：
• 用户通过客户端向YARN提交应用程序。

2. ApplicationMaster启动：
• ResourceManager分配一个 Node Manager，启动一个 Container 并运行 Application Master 实例；

3. 资源请求和分配
• ApplicationMaster启动后，向ResourceManager注册，并根据实际需要向 ResourceManager 请求更多的 Container 资源（如果作业很小, 应用管理器会选择在其自己的 JVM 中运行任务）；
• ResourceManager根据调度策略分配资源（Container）给ApplicationMaster。

4. 任务执行：
• ApplicationMaster与NodeManager通信，在分配的Container中启动任务。

5. 任务监控和故障处理：
• ApplicationMaster监控任务的执行状态，处理任务的失败和重试。
• NodeManager负责监控Container的资源使用和任务执行情况，定期向ResourceManager汇报。

6. 应用程序完成：
• 应用程序完成后，ApplicationMaster向ResourceManager注销，释放所有资源。
• ResourceManager更新应用程序状态，并将结果返回给客户端。

YARN部署配置

• 配置：yarn-env.sh文件
在文件的开头加入如下环境变量设置

# 设置 jdk 路径
export JAVA_HOME=/export/server/jdk
# 设置Hadoop路径
export HADOOP_HOME=/export/server/hadoop
# 设置Hadoop配置文件路径
export HADOOP_CONF_DIR=$HADOOP_HOME/etc/hadoop
# 设置Hadoop日志路径
export HADOOP_LOG_DIR=$HADOOP_HOME/logs

• 配置：yarn-site.xml文件
注意：其中的 node1 需要改成自己的主机用户名

<configuration>
  <!-- 
  配置项名称：yarn.log.server.url
  含义：指定YARN日志服务器的URL，用于访问作业历史记录的日志。
  参数：http://node1:19888/jobhistory/logs（作业历史日志服务器的URL）
  -->
  <property>
    <name>yarn.log.server.url</name>
    <value>http://node1:19888/jobhistory/logs</value>
  </property>

  <!-- 
  配置项名称：yarn.web-proxy.address
  含义：指定YARN Web代理的地址。
  参数：node1:8089（Web代理地址）
  -->
  <property>
    <name>yarn.web-proxy.address</name>
    <value>node1:8089</value>
  </property>

  <!-- 
  配置项名称：yarn.log-aggregation-enable
  含义：是否启用日志聚合。启用后，应用程序日志将被聚合到HDFS中。
  参数：true（启用日志聚合）
  -->
  <property>
    <name>yarn.log-aggregation-enable</name>
    <value>true</value>
  </property>

  <!-- 
  配置项名称：yarn.nodemanager.remote-app-log-dir
  含义：指定节点管理器保存远程应用程序日志的目录。
  参数：/tmp/logs（远程应用程序日志目录）
  -->
  <property>
    <name>yarn.nodemanager.remote-app-log-dir</name>
    <value>/tmp/logs</value>
  </property>

  <!-- 
  配置项名称：yarn.resourcemanager.hostname
  含义：指定YARN资源管理器的主机名。
  参数：node1（资源管理器主机名）
  -->
  <property>
    <name>yarn.resourcemanager.hostname</name>
    <value>node1</value>
  </property>

  <!-- 
  配置项名称：yarn.resourcemanager.scheduler.class
  含义：指定资源管理器使用的调度器类。
  参数：org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.FairScheduler（公平调度器类）
  -->
  <property>
    <name>yarn.resourcemanager.scheduler.class</name>
    <value>org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.FairScheduler</value>
  </property>

  <!-- 
  配置项名称：yarn.nodemanager.local-dirs
  含义：指定节点管理器的本地目录，用于存储临时文件。
  参数：/data/nm-local（本地目录）
  -->
  <property>
    <name>yarn.nodemanager.local-dirs</name>
    <value>/data/nm-local</value>
  </property>

  <!-- 
  配置项名称：yarn.nodemanager.log-dirs
  含义：指定节点管理器的日志目录，用于存储日志文件。
  参数：/data/nm-log（日志目录）
  -->
  <property>
    <name>yarn.nodemanager.log-dirs</name>
    <value>/data/nm-log</value>
  </property>

  <!-- 
  配置项名称：yarn.nodemanager.log.retain-seconds
  含义：指定节点管理器保留日志文件的时间（以秒为单位）。
  参数：10800（保留日志的时间，单位为秒）
  -->
  <property>
    <name>yarn.nodemanager.log.retain-seconds</name>
    <value>10800</value>
  </property>

  <!-- 
  配置项名称：yarn.nodemanager.aux-services
  含义：指定节点管理器的辅助服务。
  参数：mapreduce_shuffle（MapReduce Shuffle服务）
  -->
  <property>
    <name>yarn.nodemanager.aux-services</name>
    <value>mapreduce_shuffle</value>
  </property>
</configuration>

• 启动服务
• 集群启停（在主节点执行）

# 启停 yarn
start-yarn.sh
stop-yarn.sh

# 启停历史服务器
mapred --daemon start historyserver
mapred --daemon stop historyserver

• 启动流程
• 从yarn-site.xml中读取配置，确定ResourceManager所在机器，并启动它
• 读取workers文件，确定机器，启动全部的NodeManager
• 在当前机器启动ProxyServer（代理服务器）
• 单独启停（在对应的主机上执行）

yarn --daemon start resourcemanager
yarn --daemon stop resourcemanager

yarn --daemon start nodemanager
yarn --daemon stop nodemanager

yarn --daemon start proxyserver
yarn --daemon stop proxyserver

YARN的优势

• 资源利用率高：通过动态资源分配，提高集群资源的利用率。

• 扩展性好：支持多种分布式计算框架，如MapReduce、Spark、Tez等。

• 高可用性：通过ResourceManager和NodeManager的心跳机制和容错机制，保证任务的可靠执行。

• 灵活调度：支持多种调度策略（如容量调度器和公平调度器），满足不同的资源使用需求。

运行分布式计算

• wordcount：单词计数程序
• 功能：统计指定文件内各个单词出现的次数
• 给定数据输入的路径（HDFS）、给定结果输出的路径（HDFS）
• 将输入路径内的数据中的单词进行计数，将结果写到输出路径
• 程序位置：$HADOOP_HOME/share/hadoop/mapreduce/hadoop-mapreduce-examples-3.3.1.jar
• 启动命令：

语法：
hadoop jar 程序文件路径 java类名 [输入数据路径] [输出数据路径]

示例：
hadoop jar /export/server/hadoop/share/hadoop/mapreduce/hadoop-mapreduce-examples-3.3.6.jar wordcount hdfs://anjhon:8020/input hdfs://anjhon8020/output/wc

注意：
- 参数wordcount，表示运行jar包中的单词计数程序（Java Class）
- 参数1是数据输入路径（hdfs://anjhon:8020/input/wordcount/)
- 参数2是结果输出路径(hdfs://anjhon:8020/output/wc)， 需要确保输出的文件夹不存在

• 在 hdfs 中查看结果

hdfs dfs -cat /output/wc/part-r-00000

• 提交程序后，可以在YARN的WEB UI页面看到运行中的程序（http://node1:8088)



查看日志（如果点击日志后提示进不去，那么可能是网络代理的问题，关闭 VPN）

• pi：求圆周率：通过蒙特卡罗算法（统计模拟法）求圆周率
• 启动命令

语法：
hadoop jar 程序文件路径 java类名 运算节点数 计算样本数

示例：
hadoop jar /export/server/hadoop/share/hadoop/mapreduce/hadoop-mapreduce-examples-3.3.6.jar pi 3 1000