分布式计算 第五章 大数据多机计算：Hadoop

分布式计算 第五章 大数据多机计算：hadoop

5.2 大数据和分布式的基础概念

5.2.1 从硬件思考大数据

从硬件角度看，一台或是几台机器似乎难以胜任大数据的存储和计算工作。
• 大量机器的集群构成数据中心
• 使用高速互联网络对大量机器进行连接以确保数据传递
• 综合考量数据中心的散热问题、能耗问题，以及各方面成本
• 集群中硬件发生故障的概率很高，如何确保可靠性
• 单一架构的机器难以胜任各种计算类型，考虑异构计算

可靠性问题：

由于数据中心的集群中往往包含数以万计的计算机，为顾及成本，集群往往使用较为廉价的普通商用硬件

5.1.2 从软件角度看大数据

从软件角度看，由于大数据的计算与存储分布在未必可靠的大量计算机组成的集群上。

• 分而治之，使用分片存储策略和分布式算法对大数据进
行存储与处理
• 考虑存储与计算的容错性，以使得故障发生时造成的损
失最小化
• 算法设计方面要尽可能减少节点间通信（因为这很耗时）

5.1.3 分布式 distributed

谈到大数据就不得不谈到分布式，在单机上存储和处理大数据是不可能的。分布式，就是将任务分配到许多节点（机器）上去，这是一种借助网络而产生的并行方法。在大数据方向，可以将它的研究分为两方面：

• 分布式存储（如分布式文件系统、分布式数据库）
• 分布式计算（本质上是并行计算模型与算法，如上一章的mpi可以用于分布式计算）

5.1.4 谷歌“三驾马车”

在分布式领域，谷歌公司提出了三项技术来分别解决文件存储、结构化数据存储和分布式计算模型这三个关键问题，并以此开
启了大数据时代。
gfs (sosp2003)
google file system（gfs）是一个可扩展的分布式文件系统，适用于大型分布式数据密集型应用程序。它能够在廉价的商用硬件上提供容错能力，并能为大量客户端提供较高的总体性能

• 把一个较大的文件切分成不同的单元块；
• 把每个单元块存储在一个chunkserver服务器节点上，并且每一块都会复制在多个chunkserver服务器；
• 每一个文件包含多少块和哪些块，这些元数据存储在gfs master服务器上；
• 构成一个低成本的分布式存储系统，被用来处理数据量非常大的存储场景，为mapreduce的大数据处理模型提供输入和输出的存储系统；

bigtable (osdi2006)
bigtable 是一个分布式存储系统，用于管理结构化数据，旨在扩展到非常大的规模：数千个商用服务器上的 pb 级数据。 google 的许多项目（60）都将数据存储在bigtable 中，包括网络索引、google 地球和 google
财经

mapreduce (osdi2004)
mapreduce 是一种用于处理和生成大型数据集的分布式运算程序编程模型和相关实现。用户指定 map 函数来处理键-值对（kv）以生成一组中间键-值对，以及 reduce 函数来合并与同一中间键关联的所有中间值（相当于分组合并）最后得到结果。

5.1.5键值存储的优缺点

优点

• 简单：数据结构中只有键和值，并成对
  出现，值在理论上可以存放任一数据，
  并支持大数据存储。
• 快速：以内存运行模式为主，数据处理
  快是其最大优势。
• 高效计算：数据结构简单化，数据集之
  间的关系简单化，再加上基于内存的数
  据集计算，分布式计算等，形成了高效
  计算的前提条件。
• 分布式处理：分布式处理能力使键值数
  据库具备了处理大数据的能力。

缺点

• 对值进行多值查找功能很弱。
• 缺少约束容易出错。
• 不容易建立复杂关系。
• 根据其缺点可以看出来，很多查询，排序，统计等功能需要程序员在业务代码进行编程约束。

5.2 hadoop

5.2.1 hadoop与云计算区别与联系

hadoop与云计算区别
• 范围与定位
• 资源利用率
• 可靠性
hadoop与云计算联系
• 互补性
• 数据共享与交互

5.2.2 hadoop 是什么？

• hadoop 是一系列开源软件的集合，是为大数据的处理而设计的分布式框架。
• hadoop 将单机扩展到数千台机器，每台机器都提供本地计算和存储。
• hadoop 通过对应用层故障的检测和处理，再由不可靠硬件构成的集群上实现高可用。

hadoop包含了谷歌“三驾马车”的开源实现：

• gfs – hdfs
• mapreduce – hadoop mapreduce
• bigtable – hbase（不展开）

以及其他的相关组件：
• hadoop common：支持其他 hadoop 模块的通用实用程序
• hadoop yarn：作业调度和集群资源管理的框架
• ……
hdfs具有一些假设：
• 硬件故障经常发生，因此需要能检测故障并快速恢复
• 面向流式数据访问，为批处理而非用户交互使用，更注重高吞吐而非低延迟，因此并未兼容posix
• 针对大型数据集，典型文件大小为gb到tb级，不适合小文件读取，并应当在数百个节点上支持数千万的文件
• 简化的一致性模型，一个文件一旦创建、写入和关闭就不需要更改，除了追加和截断，这样简化了一致性问题且提高了吞吐
• 移动计算而非移动数据，尤其当数据集很大时，这将会较少网络拥塞并提升吞吐
• 跨软硬件平台的可移植性

5.2.2 hdfs：分布式文件系统

• hdfs是一种依照gfs设计的分布式文件系统
• 运行在低成本商业硬件上，提供高容错性
• 提供高吞吐量访问，支持具有大量数据集的应用程序
• 运行在用户态，并非内核级文件系统

namenode 和 datanode
• hdfs使用master/slave架构
• 集群包含一个namenode和多个datanode

• datanode：在hdfs中，文件被分为一到多个块（block）。datanode用于实际存储这种“块”，并处理块的创建、删除。同时，还负责处理来自文件系统客户端的读取和写入请求。

• namenode：每个集群一个（也可以有备份），用于维护文件系统的元数据（命名空间），执行文件系统命名空间上的操作，如打开、关闭、重命名文件和目录，以及确定块（block）和datanode的映射。

数据的复制
• hdfs要在大型集群中可靠地存储很大的文件。因此它将文件分块，并为每个块生成多个副本
• 每个文件的块大小和副本数量是可配置的
• namenode 做出有关块复制的所有决定。它定期从集群中的每个 datanode 接收 heartbeat 和 blockreport。收到心跳意味着 datanode 运行正常。blockreport 包含datanode 上所有块的列表

• 对同一个机架内的访问速度要快过跨机架访问，但机架也会出现故障，有一定可能会整个机架一起挂掉。因此如何放置副本对集群性能与可靠性有很大影响：
  • 全放在一个机架上，机架挂掉数据就访问不到了
  • 全放在不同机架上，写操作成本就变得很高
• hadoop拥有rack awareness（机架感知）功能，通过它可以 制定不同的副本放置策略
• 例如副本数为3时，选择写操作所在机架放置一个副本，另选一个机架放置两个副本
• 类似地，在进行读取时也会优先选择相同机架上的副本

5.2.3 yarn：调度器

• 在hadoop中，计算是以作业（job）的形式发布，并被划分为任务（task）的形式执行
• 计算任务的执行需要使用空闲计算资源（cpu等硬件资源）
• yarn就是用来调度管理计算任务和计算资源的框架

yarn的主要工作包括

• 接受作业（应用程序），启动作业，失败时重启作业
• 管理集群资源（基于容器，囊括了内存、cpu、磁盘、网络等）
• 创建、管理、监控各节点上的容器（任务执行的微环境）
• 为应用程序按需分配资源（将任务发放到适当的节点、适当的容器）
• 跟踪任务的执行，监控任务健康状态，处理任务的失败

5.2.4 mapreduce：编程模型框架

• mapreduce（后文称mr）是一个用于编写并行大数据处理程序， 并使其在集群上可靠运行的编程框架
• mr操作的数据存储在分布式文件系统上，也就是在磁盘中（在hadoop上就是hdfs）
• mr的输入输出数据形式均为键值对 <k, v>
• 一个mr作业通常将数据分为多个部分，每个部分分别由map操 作生成中间值，然后由reduce操作对具有相同key的所有value进行汇总

map与reduce操作的是数据分片而非所有数据，因此在各机器、各分片上的操作是并行的
• 数据被切分成块
• 在各机器(map worker)上启动代码副本，执行map操作，读入分块并输出中间值
• （对中间值按key进行排序）
• 在各机器(reduce worker)上启动代码副本，执行reduce操作，读入各自key对应的中间值并生成结果

reduce分为三个阶段：

• 框架首先将所有相关的map输出分片取回
• 框架将取回的中间值按key进行分组、排序
  • 以上两步是同时进行的，取回数据时就会进行合并操作
  • 可以使用job.setgroupingcomparatorclass(class)、
  job.setsortcomparatorclass(class)控制分组和排序
• 以上操作结束后，开始进行reduce

推测执行

• 在集群上分布式执行任务时，总会有一些节点跑得比其他节点慢很多
• hadoop默认情况下不会傻傻等着慢节点运行完，如果发现有的任务执行比平均速度慢，它会尝试开启一个与该任务相同的“推测任务”
• 原任务和推测任务谁先跑完就用谁，另一个会被终止
• 推测执行会占用更多集群资源，可以通过配置将其关闭