分布式计算（主讲人：强琦）在线视频

伴随着近现代硬件的发展

分布式计算作为软件系统

也有着它与之匹配的演进

我们从以下重点的

几个节点开始讨论

那么 这篇论文

已经非常有名了

如果还不知道

没有看过这篇论文的同学

自己应该好好反省一下

回去抓紧自己看

那么 我要列出来的目的

不是要跟大家讲这篇论文

而是今天在分布式

尤其是工业界 产业界

分布式计算蓬勃发展

有赖于Google的这两篇论文

第一篇是GFS

那么 当然对应着

很快就对应着开源实现 HDFS

后面开源社区的

蓬勃发展

在某种程度上

也给国内的技术从业者

提供了很大的学习的机会

和学习的空间

后来我们才可能

有机会赶上

那么 GFS主要是进行

大文件或者快文件的存储

当然 在GFS之上的系统

可以把小文件合并成大文件

然后进行存储

那么 流式文件同样

根据它的写入的特征特点

和它读取的特点

在大文件上

可以封装出流式文件

那么 GFS里头

一个很重要的特征

是它的不可修改性

当然 在这样的不可修改上

可以封装出

Mutable的功能特性

比如说LSM

我们可以支持随机读写

其扩展性可以达到跨核心

跨IDC

甚至向GFS后续的发展

可以做到跨国

那么 GFS使用

廉价的服务器 普通的服务器

降低成本

来通过replication

来增加它的可靠性

当然 在学术界和业界

后面采用再生码等技术

在大幅降低复制成本的

情况下

不降低可靠性

这也是一个值得关注的方向

跳过存储 我们现在来看编程

在此之前 用户

普通的技术开发者

很难去在集群上

编写分布式程序

和编写并行程序

并行程序和分布式程序

都有它很高的门槛

编程模型在其表达能力方面

从RPC到MPI到SQL

它的表达能力是在

不断的降弱

但是它的应用型

是在不断的增强

那么 我们既然说

要在分布式上编程

那肯定涉及到

一个核心的问题

就是你的程序

怎么能做到扩展性

第一个问题

是如何切分任务

当然 我们是分布式程序

所以 你如何在网络

根据网络拓普来切分任务

根据网络拓普

来切分任务的同时

我怎么利用

数据存储的本地化

来避免不必要的网络传输

和网络带宽

来优化我整个执行的任务

那么 有了扩展性问题

就要去解 如何去容错

大家都很清楚

假设一台机器的可靠性

是4个9的话

100台 1000台

整个集群 机器越多

每时每刻出问题的概率

几乎是百分之百

大家可以通过这个公式

可以看得很清晰

很简单因为1000台集群

1000台服务器的集群

只要有任意一个有故障

都算做整个集群有故障

然后 如何避免在很小的

在单点上的故障

引发全面的故障

也就是雪崩

大家感兴趣可以去查

前年的一些新闻

亚马逊出现了

两次比较大的故障

就是因为设计中的

一些雪崩欠缺考虑

另外 我们要去考虑到

长尾效应

什么意思呢

无论是异购机型

还是异购的网络拓普

还是根据数据分布的data scale

都会造成长尾效应

也就是说 计算会被

其中最慢的那个节点

所拖了后腿

长尾 会对整个用户的

提交任务的延时

和对雪崩

都会造成非常大的伤害

计算逻辑和编程模型

表示方面

很明显ETL的工程师

大多是希望使用SQL

建模的工程师

他们也会使用SQL

包括OLAP

而机器学习

由于其对性能的极致要求

用户会希望系统

开放最底层借口

那么 前面我们也介绍过

实际上表示能力和自由度

其实是一对treadoff

也就是说

我们今天表示能力越强

实际上这个约束就越少

而约束少

它在被分布式环境下使用

并且系统可以做到

容错的可能性会越小

如果我们对原语进行约束

那么 系统就会做

很多failover

谈到计算不可避免的

大家可以去看这篇论文

这篇论文同样

是与GFS划时代的论文

MapReduce正是约束了

用户的编写的能力

约束了编写的范式

使得系统可以做

非常多的分割

容错的工作

使得用户只关心

它的业务逻辑

当然 MapReduce是适用于

通用的数据密集型编程模型

如果你的任务涉及的

数据量不多

但是呢 大量的计算

比如说动画的渲染

那么 可能更适用于

MPI上的计算模型

当然 在MapReduce这篇论文

提出的

在2004年osdi发表的时候

论文上写每天处理20PB数据

当然 我觉得现在这个数字

应该翻了几百倍都不止

那么 Hadoop社区

随着这两篇文章的问世

开源 Hadoop就开始逐渐的

发起和兴盛

那么 最新的Hadoop 2.0

其根基在于一个系统

叫做Yarn

后面我会简单介绍一下

那么 随着MapReduce的涌现

不同的数据密集型的

编程模型也不断的涌现

例如 图计算模型Pregel

现在比较火的Spark

Dremel Drill

MapReduce的

编程模型非常简单

它把Data Type

建模成Key-value模型

那么Map function接受一条数据

返回一个list(k,v)

这个list(k,v)

会通过一个本地的Combiner

Combiner接受k,list(v)

返回list(k,v)

那么 进行这样本地的

重新组合了以后

通过网络的shuffle

进行重新组织

数据的重新组织

当然 是根据K来组织的

Reduce function

接受k, list(v)

输出list(out)

所以 大家可以看到

Reduce是接收数据

按照K重新组织以后的

一组数据来进行输出

大家可以看到这样的

编程模型非常简单

但是 它对用户写程序

进行了一定的约束

因为有了约束

所以 系统可以做非常多的

系统相关的东西

而与逻辑无关

那么 大致的执行过程

提交任务后

Master会将数据进行

split 切分

然后分配给不同的worker

也就是Map节点

每个Map节点回调

用户的Map的implement

每个worker会回调

用户的Map实现

每个Map实现

都会对着N个Reduce

通过shuffle输出文件

本地shuffle会进行sort和merge

进而可以调用Combiner

减少不必要的网络传输

当每个Map结束

它的N个Reduce的结果

都已经存储在本地

这时Master知道Map

都已经结束

拉起Reduce节点

并且告知Reduce节点的数据

Reduce的shuffle

会根据相应的地址

会去把属于它的

每个Map里面属于它的文件

拖回到本地

然后 不同Map节点

相同Reduce节点的文件

进行sort merge

sort merge完毕后

调用用户的Reduce方式

最后输出

当然 大家可以看到

里面有很多问题

就是MapReduce模型

每步都要落地

并且shuffle

是N×M的

N假设是Map

M是Reduce数

这里头存在大量的文件寻道

和传输

当然 Google MapReduce的

历史意义是非常重大

它可以自动将任务

切分成多个小的Task

而根据调度策略

又兼顾GFS本身的本地性

比如说刚才的

有些Split就在哪些机器

那么 Master在启动

Map的worker的时候

就启在那些机器上

当Task结束后

进行load balance

也就是说

所有的进程资源回收

当所有的Map结束后

根据Reduce并发

拉起进程

进行shuffle sort merge

当本节点的shuffle

全部结束后

进行Reduce task

如果task失败

它就简单的重新运行task

大家可以看到

所有的数据都是Immutable

所以 不存在版本问题

不存在一致性问题

如果节点挂了

挑选其他节点

重新运行Task

当然 这里头就要提一下

MapReduce的文章

开章就提到了

适用于MapReduce的

function的实现

必须具备确定性

否则failover无法保证

任务的正确运行

当然 刚才也提到了

由于各种原因

一定会有最慢的

和慢的那些task

那么 Reduce会启动

backup的task

会去部署到其他的机器上

最终以快的任务结束为准

当然 我们刚才也看到了

第一代的MapReduce

有非常多的问题

第一个问题 HA的问题

无论是hdfs还是MapReduce

其Master都是单点

一旦宕机会有较长的

恢复时间

或者不可恢复性

而计算的Jobtracker

全局唯一

如果有大量的任务

使用大量的文件

则会加剧元数据的膨胀

那么 资源分配

和任务内调度混合

也就是说 在Hadoop1.0

或者第一代的MapReduce

分配资源和资源分配后

任务如何摆放

是在一个决策内完成

这很明显是不合理

而第一代分配资源

是以slots

这么粗糙的粒度在做分配

它不能精细化到具体的

CPU和memory

所以 一方面造成了大量的

不够 资源

一方面 造成了大量的浪费

那么 从资源角度我们说到了

静态资源

无法进行动态的划分

一旦分配出去

这个资源无法与别人共享

也无法抢占

以slots

为粒度进行划分

划分粒度较大

隔离性较差

规模 鉴于以上原因

规模很难线性扩展

所以大家很清楚Hadoop1.0

为什么做到几千台做不上去

那么 这样子的计算

这样子的调度

这样子的架构

是能适应单一计算

也就是说

它只能跑一类MapReduce

其他类型的任务无法共用

而且 单一Master

造成整个升级非常困难

这是Hadoop2.0

也就是Yarn的架构

可以看到最明显的一个标志

即将一层资源调度

和二层业务逻辑摆放

就是worker摆放

分成了两层

分成了两个角色

也就是说RM

与AM的分离

第二点

增加了Container

保证更精细化资源的调度

比如说CPU memory

它的隔离

当然 因为它的reuse 大家知道Tez是允许Container reuse的

就不用重新拉起了

可以大幅节约

整个任务在准备阶段的时间

当然 现在的

Hadoop软件栈丰富了很多

从最底下的HDFS

到上面的Yarn

也出来了很多新的东西

比如说Tez

感兴趣的同学

大家可以去网上查资料

这个资料是非常丰富的

那包括 Slider

这个我在流计算的

分享中提到

长进程调度

和长进程的抢占隔离

长进程在Hadoop社区

现在Slider

正在迅速发展

当然 我们这门课

叫内存计算

Spark是引爆内存计算的

很重要的原因

当然了

后面我可以跟大家讨论

什么叫做内存计算

现在没有一个很狭隘的

或者很well-defined的定义

我们简单的

从分布式计算来看

你可以作出如下的抽象

第一步你如何去切任务

就是一个大的任务的DAG

你如何切分成一个一个的

stage 一个一个的task

第二个 你如何去选资源

就是你要去选足够的CPU

足够的memory

选择足够的IO

选足够的本地化 等等

找到承载这个任务的载体

第三 你要在这些载体里面

摆放以最大化运行效率

和最大化资源利用率

大家不要看这个摆

即便相同的资源

不同的摆放

整个运行效益差异几倍以上

当我选好资源 摆放好

做好谁应该运行在哪个里面

那么 进而下一步就是

我怎么下发我的执行任务

当我的任务运行的时候

我怎么去控制

它的什么时候

哪个节点该运行

哪个节点该结束

哪个节点该去做failover

所以 从整个计算来看

无法就是切 选

摆 做 控

是去重后出现了

Hive

当然了 它是facebookk开源的

大家可以看到这个简单的

一句SQL语句

HIVE来说

它是在总体上是

工作在MapReduce基础上

由Client 或者JDBC

接受用户的SQL的请求

然后 将SQL进行语法的Parser

经过逻辑执行计划

物理执行计划 下发

将一个SQL

翻译成MapReduce的

物理执行计划

下发到MapReduce机群

当然 这个已经是

一个MapReduce

Module组成的DAG 有向无环图

大家可以看到

HIVE是一个SQL的语法

所以 可以提取出大量的

元信息

进行global的存储

进行权限控制

这个图描述了

整个HIVE的

下发执行逻辑

用户提交query

发布SQL

通过编译 提交编译模块

拿到物理执行计划

然后 通过执行引擎

执行引擎相当于MapReduce

或者Hadoop的gateway

通过执行引擎

向job tracker提交任务

那么 这时候看到的

其实就是MapReduce的Job

当 MapReduce的Job

做完了以后返回

然后 执行引擎拿到结果

最终 根据用户的

DDL信息存储

并且通知用户fetch结果

所以 可以看到清晰的

HIVE是完全架构在

Hadoop的基础上

MapReduce的基础上

那么 刚才提到了

MapReduce的工作的

层次太低

而SQL leval太高

当然 它的表达能力很强

整个的MapReduce执行

是一个串行的运行

而这个DAG每一步

都会落在磁盘上面

产生大量的磁盘IO

网络IO和磁盘寻道

当然 Hadoop引入了

Distributed cache

可以将一些较小的维表

或者共用的数据存储

另外 MapReduce编写程序

对整体存储耦合过重

所以MapReduce编写代码

成本较高

那么 我们在返回SQL

来看一下

那么 SQL完全是

面向用户的视角

因为传统数据库

大量使用SQL

所以 其受众面

和用户群非常广

易用性非常好

已经有非常好的行业标准

而且不断的在演进

最新的行业标准

支持时序SQL

temporal SQL 支持graph SQL

这些都在不断的演进

那么 从系统s视角来说

schema本身就是在

理解用户的数据

可以说schema是一种知识

而且 如果你有了schema

你就会有约束

比如说这个值

比如说这个男女这一列

那它的值肯定是枚举的值

所以 你就可以把数据质量

控制在进入环节

就是进入你系统的时候

你就可以

检查它的数据schema

而不是说数据

垃圾数据 各种残缺数据进来

在查询的时候发现异常

因为我有了schema

所以 我可以制定很detail

的

数据安全策略

包括行的授权

列的授权 等等

那么 我的数据的粒度

可以到行

也可以到列

如果没有schema

你是完全无法理解

用户的数据

因为有了SQL

我们可以把

基础算子进行组合

而不像MapReduce

用户有类似的需求

它也只能做到代码级处理

不能做到直接组合

另外 SQL的出现

使得系统可以去理解

用户的功能目的

作出大量的优化

比如说计算下推

比如说无效的

列的裁减

因为你可以理解用户的功能

所以你可以做优化

而MapReduce

你完全无法理解用户的代码

另外 有了schema

我们就可以做链式压缩

这样可以大大降低

整个数据存储的成本

分布式计算领域

也有之前按照DB的思路

来做分布式计算的

比如说MPP数据库

它的架构一般都分为

存储引擎

上面是物理执行引擎

再上面是SQL

基本上都是这么三层结构

那么 当然

传统数据库

也是这么三层结构

而我们反过头来看

BigData的分层

它一般最底下也是存储层

但是这个存储层

跟DB的存储层有一些不一样

这个存储层是分布式存储层

那么 BigData一般会存在

物理执行引擎

这个物理执行引擎

定义了最基本的原语

上面的算子表示层

可以利用这个原语

实现各种不同的基础算子

比如说distinct 比如说discounting 比如说count sum

以及各种UDF UDEF

那么 在此之上

可以构建SQL的语意

可以构建machine learning的语意

可以构建特定场景语言的

一个语法

大家可以看到

DB向上只提供SQL

而BigData 向应用

即暴露物理执行原语

也可以暴露表示层原语

也可以支持各种DSL

那么 无论是DataBase

还是BigData

尤其是DataBase

发展了几十年

有相当多的理论基础沉淀

和工业实践

DB和BD在技术层面上

肯定存在融合互相借鉴

那么 很明显

BigData的优化

Schema在DB领域

逻辑执行计划

物理执行计划

CPU框架

基于成本的优化

基于规则的优化

基于历史的优化

包括Index

更精细化的存储格式

元数据

数据库领域的物化视图

内存的有效使用

内存池 对象池

等等

其在时效性上面的优化

这些都值得BigData

技术栈的学习

我在PPT刚开始就提到

BD目前大部分

都是基于scan

但是 scan并不代表着BD

Index在后面

势必会越来越多的

出现BigData的技术栈里

BigData的技术站

也会越来越实时

DB与BD的融合

已经开始发生 无法阻挡

数据库领域也会学习

分布式计算

进行 支持更复杂的

更多变的数据结构

支持嵌套结构

支持更复杂的计算

抽象出更多的表示层

开放出更多的表示能力

我们不希望在异构系统之间

拖动数据

增加数据成本

和移动成本

存不存在统一的执行器

使得不同的计算

架构在一套执行器上

而使得不同的计算

可以在运行时

复用一套数据

一套元数据

当然 前面也说到了

GFS和MapReduce的出现

Hadoop社区蓬勃发展起来

Tez Dremel Drill

Lmpala stingre Hana

Percolator Piccolo

Spark Flink相继问世

分布式计算领域系统

层出不穷