大数据技术原理与应用 - (9). Hadoop 的优化与发展

【第三篇】 - 大数据处理与分析, 《大数据技术原理与应用, 林子雨》

本篇介绍大数据处理与分析的相关技术，包括

• 第7章 - MapReduce
• 第8章 - Hive - 基于 Hadoop 的数据仓库
• 第9章 - Hadoop 的优化与发展
• 第10章 - Spark
• 第11章 - 流计算
• 第12章 - 图计算

介绍 Hadoop 2.0 对 1.0 不足与局限的解决方案，介绍 Hadoop 2.0 的新特性以及新一代资源管理调度框架 YARN 框架。

Hadoop 局限与改进

Hadoop 的局限与不足

Hadoop 1.0 的核心组件 (仅指 MapReduce 和 HDFS，不包括 Hadoop 生态系统内的 Pig、Hive、HBase 等其他组件)，主要存在以下不足：

• 抽象层次低。有时候为了实现一个简单的功能，也需要编写大量的代码。
• 表达能力有限。MapReduce 把复杂的分布式编程工作高度抽象到两个函数上，即 Map 和 Reduce，但是实际生产环境中的一些应用是无法用简单的 Map 和 Reduce 来完成。
• 开发者自己管理作业（Job）之间的依赖关系。一个 Job 只包含 Map 和 Reduce 两个阶段，通常的实际应用问题需要大量的 job 进行协作才能顺利解决，这些 job 之间往往存在复杂的依赖关系，但是 MapReduce 框架本身并没有提供相关的机制对这些依赖关系进行有效管理。
• 难以看到程序整体逻辑。
• 执行迭代操作效率低。对于一些机器学习、数据挖掘任务，往往需要多轮迭代才能得到结果，如遗传算法。但是 Map、Reduce 的过程需要对 HDFS 的数据进行读写，反复读写 HDFS 文件中的数据，大大降低了迭代操作的效率。
• 资源浪费（Map和Reduce分两阶段执行）。MapReduce 的框架设计中，Reduce 任务需要等待所有 Map 任务都完成才可以开始，造成了不必要的资源浪费。
• 实时性差（适合批处理，不支持实时交互式）。

两方面改进

• 一方面是 Hadoop 自身两大核心组件 MapReduce 和 HDFS 的架构设计改进
• 另一方面是 Hadoop 生态系统其它组件的不断丰富，加入了 Pig、Tez、Spark 和 Kafka 等新组件

改进 Hadoop 框架

完善 Hadoop 生态

组件	功能	解决 Hadoop 中存在的问题
Pig	处理大规模数据的脚本语言，用户只需要编写几条简单的语句，系统会自动转换为MapReduce作业	抽象层次低，需要手工编写大量代码
Spark	基于内存的分布式并行编程框架，具有较高的实时性，并且较好支持迭代计算	延迟高，而且不适合执行迭代计算
Oozie	工作流和协作服务引擎，协调Hadoop上运行的不同任务	没有提供作业（Job）之间依赖关系管理机制，需要用户自己处理作业之间依赖关系
Tez	支持DAG作业的计算框架，对作业的操作进行重新分解和组合，形成一个大的DAG作业，减少不必要操作	不同的MapReduce任务之间存在重复操作，降低了效率
Kafka	分布式发布订阅消息系统，一般作为企业大数据分析平台的数据交换枢纽，不同类型的分布式系统可以统一接入到Kafka，实现和Hadoop各个组件之间的不同类型数据的实时高效交换	Hadoop生态系统中各个组件和其他产品之间缺乏统一的、高效的数据交换中介

HDFS2.0 的新特性

HDFS HA

HDFS 1.0 的不足

• HDFS 1.0 只存在一个 NameNode，存在单点故障问题。
• 第二名称节点 (SecondaryNameNode) 无法解决单点故障问题，非 NameNode 的热备份。

HDFS HA（High Availability）是为了解决单点故障问题，是 NameNode 的热备份解决方案。

• HA 集群设置两个 NameNode，“活跃 (Active)”和“待命 (Standby)”。
• 两种 NameNode 的状态同步，可以借助于一个共享存储系统来实现。
• 一旦活跃 NameNode 出现故障，就可以立即切换到待命 NameNode。
• Zookeeper 确保只有一个 NameNode 在对外服务。
• NameNode 维护映射信息，数据节点同时向两个 NameNode 汇报信息。

HDFS Federation

HDFS 1.0 的不足

• 单点故障问题
• 不可以水平扩展（是否可以通过纵向扩展来解决？）
• 系统整体性能受限于单个 NameNode 的吞吐量
• 单个 NameNode 难以提供不同程序之间的隔离性
• HDFS HA 是热备份，提供高可用性，但是无法解决可扩展性、系统性能和隔离性

HDFS Federation 的设计

• 在 HDFS Federation 中，设计了多个相互独立的 NameNode，使得 HDFS 的命名服务能够水平扩展，这些 NameNode 分别进行各自命名空间和块的管理，相互之间是联盟 (Federation) 关系，不需要彼此协调。并且向后兼容。
• HDFS Federation 中，所有 NameNode 会共享底层的 DataNode 存储资源，DataNode 向所有 NameNode 汇报。
• 属于同一个命名空间的块构成一个 “块池”。

HDFS Federation 的访问方式

• 对于 Federation 中的多个命名空间，可以采用客户端挂载表 （lient SideMount Table) 方式进行数据共享和访问。
• 客户可以访问不同的挂载点来访问不同的子命名空间
• 把各个命名空间挂载到全局 “挂载表” (mount-table) 中，实现数据全局共享
• 同样的命名空间挂载到个人的挂载表中，就成为应用程序可见的命名空间

HDFS Federation 相对于 HDFS1.0 的优势

1. HDFS集群扩展性。多个 NameNode 各自分管一部分目录，使得一个集群可以扩展到更多节点，不再像 HDFS1.0 中那样由于内存的限制制约文件存储数目。
2. 性能更高效。多个 NameNode 管理不同的数据，且同时对外提供服务，将为用户提供更高的读写吞吐率。
3. 良好的隔离性。用户可根据需要将不同业务数据交由不同 NameNode 管理，这样不同业务之间影响很小。

需要注意: HDFS Federation 并不能解决单点故障问题，也就是说，每个 NameNode 都存在在单点故障问题，需要为每个 NameNode 部署一个后备 NameNode，以应对 NameNode 挂掉对业务产生的影响。

YARN

MapReduce1.0 的缺陷

1. 存在单点故障
2. JobTracker “大包大揽”导致任务过重 (任务多时内存开销大，上限4000节点)
3. 容易出现内存溢出 (分配资源只考虑 MapReduce 任务数，不考虑 CPU、内存)
4. 资源划分不合理 (强制划分为 slot ，包括 Map slot 和 Reduce slot)

YARN 设计思路

• MapReduce1.0 既是一个计算框架，也是一个资源管理调度框架
• 到了 Hadoop2.0 以后，MapReduce1.0 中的资源管理调度功能，被单独分离出来形成了 YARN，它是一个纯粹的资源管理调度框架，而不是一个计算框架
• 被剥离了资源管理调度功能的 MapReduce 框架就变成了 MapReduce2.0，它是运行在 YARN 之上的一个纯粹的计算框架，不再自己负责资源调度管理服务，而是由 YARN 为其提供资源管理调度服务

YARN 体系结构

• ResourceManager:
  • 处理客户端请求
  • 启动/监控 ApplicationMaster
  • 监控 NodeManager
  • 资源分配与调度
• ApplicationMaster:
  • 为应用程序申请资源，并分配给内部任务
  • 任务调度、监控与容错
• NodeManager:
  • 单个节点上的资源管理
  • 处理来自 ResourceManger 的命令
  • 处理来自 ApplicationMaster 的命令

YARN 工作流程

1. 用户编写客户端 (Clinet) 应用程序，向 YARN 提交应用程序，提交的内容包括 ApplicationMaster 程序、启动 ApplicationMaster 的命令、用户程序等
2. YARN 中的 ResourceManager 负责接收和处理来自 Clinet 的请求，为应用程序分配一个容器，在该容器中启动一个 ApplicationMaster
3. ApplicationMaster 被创建后会首先向 ResourceManager 注册
4. ApplicationMaster 采用轮询的方式向 ResourceManager 申请资源
5. ResourceManager 以“容器”的形式向提出申请的 ApplicationMaster 分配资源
6. 在容器中启动任务（运行环境、脚本）
7. 各个任务向 ApplicationMaster 汇报自己的状态和进度
8. 应用程序运行完成后，ApplicationMaster 向 ResourceManager 的应用程序管理器注销并关闭自己

Hadoop 生态中代表性功能组件

Pig

Pig 是什么

Pig 是一个基于 Apache Hadoop 的大规模 High-Level 数据分析平台。Pig 可以 在MapReduce，Apache Tez 或 Apache Spark 中执行其 Hadoop 作业，它提供的 SQL-LIKE 语言 - Pig Latin，该语言的编译器会把类 SQL 的数据分析请求转换为一系列经过优化处理的 MapReduce 运算。Pig 为复杂的海量数据并行计算提供了一个简单的操作和编程接口，Pig Latin 可以使用 user-defined functions (UDF) 进行扩展，用户可以使用 Java，Python，JavaScript，Ruby 或 Groovy 编写这些函数，然后直接从该语言调用。

Pig 可以加载数据、表达转换数据以及存储最终结果。

Pig Latin

Pig Latin是这样的一个操作：通过对关系进行处理产生另外一组关系。Pig Latin语言在书写一条语句的时候能够跨越多行，但是必须以半角的分号来结束。

Pig Latin语言通常按照下面的流程来编写：

1. 通过一条 LOAD语句 从文件系统中读取数据；
2. 通过一系列 转换语句 对数据进行处理；
3. 通过一条 STORE语句 把处理结果输出到文件系统中，或者使用一条DUMP语句把处理结果输出在屏幕上。

LOAD和STORE语句有严格的语法规定，用户很容易就能掌握，关键是如何灵活的使用转换语句对数据进行处理。

Pig Latin 例子

example

input_lines = LOAD '/tmp/my-copy-of-all-pages-on-internet' AS (line:chararray);

-- Extract words from each line and put them into a pig bag
-- datatype, then flatten the bag to get one word on each row
words = FOREACH input_lines GENERATE FLATTEN(TOKENIZE(line)) AS word;

-- filter out any words that are just white spaces
filtered_words = FILTER words BY word MATCHES '\\w+';

-- create a group for each word
word_groups = GROUP filtered_words BY word;

-- count the entries in each group
word_count = FOREACH word_groups GENERATE COUNT(filtered_words) AS count, group AS word;

-- order the records by count
ordered_word_count = ORDER word_count BY count DESC;
STORE ordered_word_count INTO '/tmp/number-of-words-on-internet';

上面的程序将生成并行的可执行任务，这些任务可以分布在 Hadoop 集群中的多台计算机上，以计算数据集（例如互联网上的所有网页）中的单词数。

Tez

Tez 是 Apache 开源的支持 DAG作业的计算框架，它直接源于 MapReduce 框架，核心思想是将 Map 和 Reduce 两个操作进一步拆分。

• Map 被拆分成 Input、Processor、Sort、Merge 和 Output
• Reduce 被拆分成 Input、Shuffle、Sort、Merge、Processor 和 Output 等

分解后的元操作可以任意灵活组合，产生新的操作，这些操作经过一些控制程序组装后，可形成一个大的 DAG 作业。通过 DAG 作业的方式运行 MapReduce 作业，提供了程序运行的整体处理逻辑，就可以去除工作流当中多余的 Map 阶段，减少不必要的操作，提升数据处理的性能。

MapReduce 和 Tez 对比

Tez的优化主要体现在，去除了

• 连续两个作业之间的 “写入HDFS”
• 每个工作流中多余的 Map 阶段

比如以下 Hive SQL 会翻译成四个MR作业，而采用 Tez 则生成一个 DAG 作业，可大大减少磁盘 IO：

SELECT a.state, COUNT(*),AVERAGE(c.price)
FROM a
JOIN b ON (a.id = b.id)
JOIN c ON (a.itemId = c.itemId)
GROUP BY a.state

Tez 作用

• 在 Hadoop2.0 生态系统中 (参考上图: Hadoop 1.0 到 2.0)，MapReduce、Hive、Pig 等计算框架，都需要最终以 MapReduce 任务的形式执行数据分析，因此 Tez 框架可以发挥重要的作用
• 借助于 Tez 框架实现对 MapReduce、Pig 和 Hive 等的性能优化
• 可以解决现有 MapReduce 框架在迭代计算 (如 PageRank algorithms) 和交互式计算方面的问题

(Tez+Hive) 与 Impala、Dremel 和 Drill 的区别？

• Tez 在解决 Hive、Pig 延迟大、性能低等问题的思路，是和那些支持实时交互式查询分析的产品（如 Impala、Dremel 和 Drill 等）是不同的。
• Impala、Dremel 和 Drill 的解决问题思路是抛弃 MapReduce 计算框架，不再将类似 SQL 语句的 HiveQL 或者 Pig 语句翻译成 MapReduce 程序，而是采用与商用并行关系数据库类似的分布式查询引擎，可以直接从 HDFS 或者 HBase 中用 SQL 语句查询数据，而不需要把 SQL 语句转化成 MapReduce 任务来执行，从而大大降低了延迟，很好地满足了实时查询的要求。
• Tez 则不同，比如，针对 Hive 数据仓库进行优化的 “Tez+Hive” 解决方案，仍采用 MapReduce 计算框架，但是对 DAG 的作业依赖关系进行了裁剪，并将多个小作业合并成一个大作业，这样，不仅计算量减少了，而且写HDFS次数也会大大减少。

Spark

Hadoop 缺陷: 其 MapReduce 计算模型延迟过高，无法胜任实时、快速计算的需求，因而只适用于离线批处理的应用场景

• 中间结果写入磁盘，每次运行都从磁盘读数据
• 在前一个任务执行完成之前，其他任务无法开始，难以胜任复杂、多阶段的计算任务

Spark 最初诞生于伯克利大学的 APM 实验室，是一个可应用于大规模数据处理的快速、通用引擎，如今是 Apache 软件基金会下的顶级开源项目之一。Spark 在借鉴Hadoop MapReduce 优点的同时，很好地解决了 MapReduce 所面临的问题:

• 内存计算，带来了更高的迭代运算效率
• 基于 DAG 的任务调度执行机制，优于 MapReduce 的迭代执行机制

详见下一章 - 第10章: Spark 内容。

Kafka

• Kafka 是一种高吞吐量的分布式发布订阅消息系统，用户通过 Kafka 系统可以发布大量的消息，同时也能实时订阅消费消息
• Kafka 可以同时满足在线实时处理和批量离线处理
• 在公司的大数据生态系统中，可以把 Kafka 作为数据交换枢纽，不同类型的分布式系统（关系数据库、NoSQL 数据库、流处理系统、批处理系统等），可以统一接入到Kafka，实现和 Hadoop 各个组件之间的不同类型数据的实时高效交换