by 清华大学
为什么并行计算？
- 计算量大
- 单进程算得不够快，多CPU算
- 内存需求大
- 单机内存不够大
- 内存随机访问比硬盘随机访问快100,000倍
- I/O 量大
- 单个硬盘读写太慢，多个硬盘读写
并行计算的挑战
- 编程困难
- 并行性识别与表达，难写
- 同步语句，难写对
- 性能调优难，难写快 （并行计算目标就是提升性能，性能调优难）
-负载平衡- 局部性 （高速缓存cache，使用cache可以快10倍左右）
- 容错难
并行计算中的局部性
矩阵相乘，按列访问会造成cache失效
image.png
分块算法，得到更高的局部性
高可用性
大数据处理系统通常是由大量不可靠服务器组成的，如果处理1个10天的大数据处理任务时在第8天机器坏掉怎么办？
重新计算不一定能解决问题
传统的容错方法不适用
- 锁步法（性能会有较大影响），多版本编程（多个人来编程，对比结果，软件容错）
检查点设置与恢复（保存程序状态，从保存状态位置继续执行，IO量大）
大数据处理并行系统
image.png
内存计算需求
Map Reduce成功之处
- 用户只需要编写串行程序
- 自动并行化和分布式执行
- 自动容错
- 自动负载平衡
用户对系统提出了更高的要求 - 更复杂的多阶段任务
- 交互式查询
Map Reduce 的局限性 - 表达能力有限
- 只有Map 和Reduce两种操作
- 复杂任务通常需要迭代的 MapReduce
- 需要将中间结构保存在硬盘上
- 大量I/O操作造成性能急剧下降
- 引入的I/O操作多，只能做离线分析，很难支持数据的交互式查询
MapReduce 文件传递数据
image.png
如果能用内存保存数据？
image.png
比采用硬盘方案快10-100倍
In Memory Computing
内存计算的可行性
问题：
- 内存是否足够大能够装下所需要的数据？
- 内存有多贵？与硬盘相比性价比如何？
- 数据保存在硬盘上，可以保证数据的可用性，放在内存里如何容错？
- 如何高效表示内存里的数据？
input -> iter1 -> memory -> iter2 -> memory
image.png
单位芯片上集成的晶体管数量随着时间（每两年）可以成倍增长
各个内存层次的延迟
image.png
DRAM比硬盘快100,000倍，但是DRAM比片上cache慢6-200倍
内存计算的实例：SPARK
SPARK设计理念：着重效率和容错
如何抽象多台机器的内存？
- 分布式共享内存（DSM）
- 统一地址空间
- 很难容错
- 分布式键-值存储（Piccolo，RAMCloud）
- 允许细粒度访问
- 可以修改数据（mutable）
- 容错开销大
DSM和键值对的容错机制
- 副本或Log
- 对数据密集应用来说开销很大
- 比内存写要慢10-100倍
解决方案
- RDD（Resilient Distributed Datasets）
- 基于数据集合，而不是单个数据
- 由确定性的粗粒度操作产生（map,filter,join等）
- 数据一旦产生，就不能修改（immutable）
- 如果要修改数据，要通过数据集的变换来产生新的数据集
高效容错方法
- 数据一旦是确定性的产生，并且产生后不会变化
- 就可以通过“重复计算”的方法恢复数据
-
只要记住rdd生成过程就可以了，这样一次log可以用于很多数据，在不出错的时候几乎没有开销
image.png
大数据处理并行系统
用编程模型上的限制获取好的容错能力和高性能
image.png
K-V 对，细粒度修改； HDFS 只能添加数据
RDD 高吞吐率，不允许做细粒度修改，换取好的容错能力和好的性能
SPARK 编程接口
- 基于Scala 语言
- 类似Java的一种函数语言
- 可以在Scala控制台上交互式地使用Spark
- 现在也支持Java 和Python
- 基于RDD的操作
- Transformation： 从现有RDD产生新的RDD
- map, reduce, filter, groupBy, sort, distinct, sample ...
- Action: 从RDD返回一个值
-
count, collect, first ,foreach ...
image.png
- Transformation： 从现有RDD产生新的RDD
SPARK 编程实例--LOG挖掘
将数据从文件系统中调入内存，然后进行交互式的查询
image.png
lines = spark.textFile("hdfs://...")
errors = lines.filter(_.startsWith("ERROR"))
messages = errors.map(_.split('\t')(2))
cachedMsgs = messages.cache()
cachedMsgs.filter(_.contains("foo")).count // 包含foo的信息的数目
cachedMsgs.filter(_.contains("bar")).count
image.png
image.png
image.png
image.png
SPARK 实现技术
延迟估值（Lazy Evaluation）
val lines = sc.textFile("data.txt") //transformation
val lineLengths = lines.map(s => s.length) //transformation
val totalLength = lineLengths.reduce((a,b) => a+b) //action, trigger computation
- 前面两行都不会触发计算（Transformation）
- 最后一行的reduce会引发计算，生成DAG
image.png
有向无环图
image.png
image.png
image.png
image.png
RDD性能的提高
对需要重用的RDD使用Persist和Cache提高性能
image.png
image.png
SPARK 应用和生态环境
image.png
image.png
image.png
image.png
image.png
SPARK 局限性
image.png
image.png
只能复制一份，标记少数节点。操作为网络操作、内存拷贝操作、IO操作（由于数据是只读的）-> 效率低，大量内存拷贝。
每次细粒度的数据更新，由于spark基于粗粒度RDD只读的数据对象模型，需要RDD变换，即有大量数据的复制，导致处理效率不高
