【原创】《大数据互联网大规模数据挖掘与分布式处理》阅读笔记(二)

【原创】《大数据互联网大规模数据挖掘与分布式处理》第二章大规模文件系统及MAP-Reduce

首先来看看这一章讲解的整体架构，分别介绍了分布式文件系统、Map-Reduce、使用Map-Reduce的算法，Map-Reduce扩展和集群计算算法的效率问题。



一、分布式文件系统

1.1分布式文件系统起因
大规模WEB服务流行 在数千个计算节点完成大规模计算 这些计算节点都是普通的硬件构成，为了发挥并行化的优势，保证可靠性 开发专用的文件系统。

1.2分布式文件系统特点
本书写的特征是：1）存储单位比传统操作系统中的磁盘快大

2）提供数据冗余机制来防止数据分布在上千块磁盘上是频发媒介故障


联想：如果用户同时对一个文件进行写操作，如何同步，是不是采取加锁方案。

搜一搜：

如果文件的访问仅限于一个用户，那么分布式文件系统就很容易实现。可惜的是，在许多网络环境中这种限制是不现实的，必须采取并发控制来实现文件的多用户访问，表现为如下几个形式：

只读共享
任何客户机只能访问文件，而不能修改它，这实现起来很简单。HADOOP采取这种形式。

受控写操作采用这种方法，可有多个用户打开一个文件，但只有一个用户进行写修改。而该用户所作的修改并不一定出现在其它已打开此文件的用户的屏幕上。

并发写操作这种方法允许多个用户同时读写一个文件。但这需要操作系统作大量的监控工作以防止文件重写，并保证用户能够看到最新信息。这种方法即使实现得很好，许多环境中的处理要求和网络通信量也可能使它变得不可接受。

1.3计算节点的物理结构
集群计算：一种新的并行计算架构，其中计算节点位于机架，单节点之间网络互连，机架之间另一级网络或交换机互连。

主要故障模式：单节点故障；单机架故障

解决方案：文本需要多副本存储；计算过程需要分成多个任务完成。

联想：文本过多，数据冗余过渡也是一种资源浪费，判定副本存储数量的阈值如何确定？

联想：计算过程中多任务完成，其中某些任务突然因为外界原因宕机，怎么切换到其他机子上运行？

搜一搜：

l 例如在HADOOP的HDFS文件系统中，默认的副本个数是3，两个存放在一个机架，另一个放在其他机架上。在副本机制中，副本的数量值意味着对数据的存储空间的需求是原始数据的数倍，对于非重要的数据而言就是空间浪费，因此需要考虑如何基于数据的重要性来设置文本的副本数，搜到了一篇硕士论文讲述的是基于历史统计记录的动态副本策略，链接如下：http://cdmd.cnki.com.cn/Article/CDMD-10183-2010109806.htm

l 在HADOOP的HDFS中有一种心跳机制，可以掌握集群的工作状况。DataNode通过周期性地发送信条信息与NameNode联系，Namenode通过获取的心跳信息得知DataNode的存在，及其上的磁盘容量、已用空间、总负载等信息，可以参考这片博文讲解的比较详细：/article/8839090.html

l 通过一些负载管理工具进行分布式文件的管理，还可以通过快照定期存储某个时刻的数据复制，当数据损坏的时候，可以回滚到一个过去已知的正确的时间点。

二、Map-Reduce
2.1 Map任务
Map-reduce的思想就是“分而治之”, Mapper负责“分”，即把复杂的任务分解为若干个“简单的任务”执行.“简单的任务”有几个含义：

l 数据或计算规模相对于原任务要大大缩小；

l 就近计算，即会被分配到存放了所需数据的节点进行计算；

l 这些小任务可以并行计算，彼此间几乎没有依赖关系。

Map 任务的输入文件由多个元素组成，元素可以是任意类型。所有Map任务的输入和Reduce任务的输出都是键-值对的形式。此时的“键”不要求它们的唯一性，一个Map任务可以生产多个具有相同键的键-值对。

2.2 分组和聚合
分组：主控进程通常选择一个哈希函数作用于键并产生一个0到r-1（r为reduce任务的数目）的桶编号。Map任务输出的每个键都被哈希函数作用，根据哈希结果其键-值将被放到r个本地文件中的一个。每个文件都会被指派一个Reduce任务。

合并：主控进程将每个Map任务输出的面向某个特定Reduce任务的文件合并，还是以键-值对序列传给进程。

2.3 Reduce任务
对map阶段的结果进行汇总，Reducer的数目由mapred-site.xml配置文件里的项目mapred.reduce.tasks决定。缺省值为3，用户可以覆盖之。

Reduce任务的输出是键-值对序列，键Key是Reduce任务接收到的输入键，值value是其接收到与key关联的值的组合结果。

所有的Reduce任务的输出结果会合并成一个文件。

2.4 组合器
Reduce函数满足交换律和结合律，即组合值可以按照任何次序组合，其结果不变。

2.4 Map-Reduce的执行细节
一般而言，每个工作进程要么处理Map任务，要么处理Reduce任务。

主控进程要负责创建一定数目的Map和Reduce任务。一般情况下对于输入文件的每个文件都会创建一个Map任务，但是要限制Reduce任务的数量，因为每个Map任务都必须给每个Reduce任务建立中间文件，Reduce任务太多，会导致中间文件数目暴增。

主控进程要负责记录每个Map和Reduce的运行状态，分配新任务或对宕机的节点上的任务进行处理等。

2.4 节点失效的处理
两种情况，主控进程的计算节点崩溃和Reduce任务的计算节点失效，前者整个Map-Reduce作业都需要重启，后者则是需要将失效节点上运行的Reduce任务的状态置为“空闲”，并安排另外的工作节点重新运行。

联想：主控进程的计算节点崩溃就需要重启整个作业，这势必会造成巨大的损失，有没其他方法。

搜一搜：

事实上，设计者对Namenode的单点故障还是有所考虑的，比如在hadoop会设置一个Secondary NameNode，即辅助名称节点，辅助名称节点是在名称节点发生故障时替换名称节点的。SecondNamenode是对主Namenode的一个补充，它会周期的执行对HDFS元数据的检查点。当前的设计仅仅允许每个HDFS只有单个SecondNamenode结点。SecondNamenode是有一个后台的进程，会定期的被唤醒（唤醒的周期依赖相关配置）执行检查点任务，然后继续休眠。它使用ClientProtocol协议与主Namenode通信。
可以参考链接：http://f.dataguru.cn/thread-20476-1-1.htm
http://blog.csdn.net/shatelang/article/details/7595373?reload
实践一下：

有一个文本文件，里面每一行是一个单词，统计这个文件里总共有哪些不同的单词（相当于去重问题），设计Map-Reduce算法过程。

解：

假设有640M的文件，默认block大小为64M，10台Hadoop机器，使用5个reduce任务：

1. 每个map任务按行读取文件，读取的行数据交给map函数执行，通过指定的TextInputFormat一次处理一行。然后，它通过StringTokenizer以空格为分隔符将一行切分为若干tokens，之后，输出< <word>, 1>形式的键值对。

2. 如果有combiner,就对第一步的输出进行combiner操作。即每次map过程介绍之后，会对输出按照key进行排序，然后把输出传递给本地的combiner（按照作业的配置是与Reducer一样的，都使用的是Reduce.class），进行本地聚合,将相同word的计数累加。这样可以将中间结果大大减少，减少后续partitioner, sort, copy的开销，提高性能。

3. 每个map任务对自己的输出进行分区，默认的分区操作是对key进行hash，并对reduce任务数求余，这样相同的单词都被分在同一分区内。之后将对同一分区内的数据按key进行排序，使之成为各分区内有序。

4. Reduce任务各自从map机器上copy属于自己的文件，并且进行合并。合并好后进行Sort操作，再次把不同小文件中的同一单词聚合在一起，作为提供给reduce操作的数据。

5. 进行reduce操作，对同一个单词的value列表再次进行累加，最终得到每个单词的词频数。一般是调用Reducer接口中的 reduce方法,输入参数中的 key, values是由 Map任务输出的中间结果，values是一个 Iterator,遍历这个 Iterator,就可以得到属于同一个Key的所有 value。此处，key是一个单词，value是词频。只需要将所有的 value相加，就可以得到这个单词的总的出现次数。

6. 最后Reduce把结果写到磁盘。

三、使用Map-Reduce的算法

首先要明确的是Map-Reduce框架不是万能通用的，整个分布式文件系统只在文件巨大、更新很少的情况下才有意义。例如像矩阵-向量和矩阵-矩阵计算非常适合Map-Reduce计算框架。

3.1 Map-Reduce的矩阵-向量乘法实现
先要明确的是Map-Reduce框架不是万能通用的，整个分布式文件系统只在文件巨大、更新很少的情况下才有意义。例如像矩阵-向量和矩阵-矩阵计算非常适合Map-Reduce计算框架。

一个n*n的矩阵M和n维向量v的乘积是一个n维向量x。

Map函数每个Map任务讲整个向量v和矩阵M的一个文件块作为输入。对每个矩阵元素，Map任务都有一个键-值对()。

Reduce函数 Reduce任务将所有与给定键i相关的值相加即可。

3.2 向量v无法放入内存时的处理
解决方案：将矩阵分割成为多个宽度相等的垂直条，同时将向量分割成同样数目的水平条，每个水平条的高度等于矩阵垂直条的宽度。矩阵第i个垂直条只和向量的第i个水平条相乘。

3.3 关系代数运算
由于大规模数据处理势必要用到许多的数据库查询操作，这节简单回顾了查询中常用的几个关系运算:选择,投影,并,交,差,自然连接,分组和聚合.

选择:对关系R的每一个元素应用条件C,得到仅满足条件C的元组。

投影:对关系R的某一个属性子集S,从每个元组中得到仅包含S中属性的元素。

并、交、差用于两个具有相同模式的关系的元组集合上。

自然连接：如果两个元组的所有公共属性的属性值一致，就生成了一个新的元组，这个元组有原来两个元组的公共部分加上非公共部分组成。

分组和聚合：给定关系R，分组是指按照属性集合G中的值对元组进行分割。然后对每个组的值按照某些其他的属性进行聚合。典型的聚合运算：SUM、COUNT、AVG、MIN、MAX。

3.4 基于Map-Reduce的选择运算
Map: 对于每个R中的元组t，检测它是否满足C，如果满足则产生一个键-值对(t,t)，键值都是t。

Reduce: 仅将每一个键-值对传递到输出部分。

3.5 基于Map-Reduce的投影运算
Map: 对于每个R中的元组t，剔除t中属性不在S中的字段得到元组
，输出键-值对 。

Reduce: 对任意Map任务产生的每个键 ，将存在一个或者多个键-值对 ，Reduce函数将 转化成
，以保证该键值只产生一个键-值对。

3.6 基于Map-Reduce的并、交、差运算
Map只将相同模式关系R和S的输入元组作为键-值对输给Reduce任务，Reduce只需要向投影一样剔除冗余。

并

Map: 将每个输入的元组t转变为键-值对 。

Reduce: 和每个键t关联的可能有一个或者两个值，两种情况下都输出

交

Map: 将每个输入的元组t转变为键-值对 。

Reduce:
如果键t的值表为 ，则输出 ，否则输出 。

差

Map:
对于R中的元组t，产生键-值对 。对于S中的元组t，产生键-值对.R,S为关系的名称，并非关系本身。

Reduce:
如果相关联的值表为[R]，则输出键-值对 ，否则输出 。

3.7 基于Map-Reduce的自然连接运算
实例：将关系R(A,B)和S(B,C)进行自然连接运算。
Map:
对于R中的元组t，产生键-值对 。对于S中的元组t，产生键-值对.

Reduce: 基于(R,a)和(S,c)构建的所有对，键b所对应的输出结果为(b,[(a1,b,c2),
(a2,b,c2), …])

上述算法在关系多余两个属性的情况下同样适用。

3.8 基于Map-Reduce的分组和聚合运算
Map: 对于每个元组(a,b,c)产生键-值对 。

Reduce:
每一个键a代表一个分组，即对与键a相关的字段B的值表施加Θ操作，输出结果对(a,x),x为Θ操作的结果。

3.9 矩阵乘法
矩阵乘积相当于一个自然连接运算再加上分组和聚合运算。

Map: 将每一个矩阵元素mij传值给键-值对,将每一个矩阵元素njk传值给键-值对
。
Reduce: 对于来自M的键-值对 和来自N的键-值对
，产生 。
然后再通过另外一个Map-Reduce运算进行分组聚类。将上面的reduce函数结果传递给Map函数，形式为 ，产生键-值对
，最后通过Reduce，对于每个键(i,k)，将与此键关联的的所有值的和，结果计为((i,k),v)。

3.10 基于单步Map-Reduce的矩阵乘法
Map: 将每一个矩阵元素mij传值给键-值对,将每一个矩阵元素njk传值给键-值对
。
Reduce: 将两个列表的第j个元组中的 和 出相乘，然后再将这些积相加，最后于键(i,k)组对作为Reduce函数的输出结果。

四、Map-Reduce的扩展

改进和系统和Map-Reduce具有相同的共同特征
（1）建立在分布式文件系统之上
（2）管理大量任务，任务是用户编写函数的实例化结果
（3）提供大任务执行过程中发生的大部分失效的处理方法。

4.1 工作流系统
将Map-Reduce一般化为支持任意无环函数集的系统，每个函数都可以实例化为任意数目的任务。
关键在于任何任务只有在输入准备好之后才能输出。这样就可以保证如果某个任务失败，其结果不会传递给它工作流中的
后续任务，避免了通信开销。

4.2 Map-Reduce的递归扩展版本
递归主要是通过Map-Reduce过程的迭代调用实现，因为一个真正的递归任务不能独立启动失效任务。这就要求我们在存在递归工作流的系统中，引入某些机制来处理任务失效，而不是重启。

介绍了递归任务集中的连接任务和去重任务，主控进程将一直等待直到每个连接任务完成对其完整输入的一轮处理，再将输出的文件分部到查重任务作为输入，再讲输出结果作为连接任务的下一轮输入。如果每个任务都保存曾产生过的输出文件，并且把连接和去重任务放在不同机架上，就可以处理单节点故障和单机架故障。

4.3 Pregel系统
Pregel系统提供了在计算集群中实现递归算法时的处理失效任务的方法。在计算过程中设立检查点，在执行超步后记录整个计算的现场，对每个任务的全部状态进行记录，在处理失效任务时就可以从最近的检查点进行重启。

五、集群计算算法的效率问题

这节介绍了度量集群计算算法的质量模型。

5.1 集群计算的通信开销模型

什么是通信开销模型？

它是指实现该算法的所有任务的通信开销之和，通常开销是指数据从创建地到使用地的传输开销。

为什么要关注通信开销？

算法中每个执行任务一般非常简单，时间复杂度一般和输入规模成线性关系；

计算集群中的互连速度是GB/s，在任务传输的同等时间内，计算机可以做大量的工作；

若任务执行时需要的文件块存放在磁盘上，传输至内存的时间很长。

为什么仅计算输入规模还不计算输出规模？

因为任务A的输出是另一个任务B的输入，没有必要计算输出规模；

而且实际中，任务的输出比输入规模或任务中产生的中间数据相比更小一些。

特别地Map任务的输出和输入规模大体相当，Map到Reduce任务的通信一般通过集群来互连不需要内存到磁盘的传输。

5.2 实耗通信开销

实耗通信开销是指在无环网络图所有路径中最大的通信开销。

它相当于并行算法的最短执行时钟时间，可以通过将所有工作平均分配给不同的任务来使通信开销达到尽可能的小。

5.3 多路连接
实例：三个关系R(A,B)、S(B,C)和T(C,D)的一次性连接运算的总通信开销

(1)S 将每个元组S(v,w)仅仅传递一次到Reduce任务(h(v),g(w));

(2)cr 将每个元组R(u,v)传递到c个Reduce任务(h(v),y)，y可能取值c个；

(3)bt 将每个元组T(v,x)传递到b个Reduce任务(z,g(w))，z可能取值b个；

(4)r+s+t
每个关系的每个元组输入到Map任务的开销

拉格朗日乘子求解，优化每个关系的复制度。