第1章 数据分片与路由
• 主流的大数据存储与计算系统采用横向扩展（Scale Out）的方式支持系统可扩展性，通过增加机器数目来获得水平扩展能力。因此，对于待存储处理的海量数据，需要通过数据分片（Shard/Partition）来将数据进行切分并分配到各个机器中去，数据分片后，如何能够找到某条记录的存储位置称为数据路由。
• 数据分片实现系统的水平扩展，数据复制来保证数据的高可用性
• 哈希分片
○ Round Robin：Round Robin就是哈希取模法。假设有K台物理机，通过一下哈希函数实现数据分片：H（key）=hash（key）mod K
§ 优点：实现简单。但缺乏灵活性，新增一台物理机要把所有数据按改编后的函数再次分配一遍
○ 虚拟桶（Virtual Buckets）：Membase（现更名为Couchbase）是一个内存分布式NoSQL数据库，在待存储记录和物理机之间引入了虚拟桶层，所有记录首先通过哈希函数映射到对应的虚拟桶，记录和虚拟桶是多对一的映射关系，即一个虚拟桶包含多个记录信息；第二层映射是虚拟桶和物理机之间的映射关系，也是多对一的映射，一个物理机可以容纳多个虚拟桶，即Membase通过内存表管理这层映射关系。Membase的虚拟桶层就是对应的“数据分片”层，一个虚拟桶即是一个数据分片。Key-Partition映射采用哈希函数，而Partition-machine采用表格管理实现。
§
§ 与Round Robin相比，Membase引入了虚拟桶层，将原先由记录直接到物理机的单层映射解耦为两级映射，加强了系统的扩展灵活性。当新加入机器时，将某些虚拟桶从原先分配的机器重新分配给新机器，只需要修改Partition-machine映射表受影响的个别条目就能实现扩展，具有较强的灵活性。
○ 一致性哈希（consistent hashing）：分布式哈希表（DHT）是P2P网络和分布式存储中厂常用的技术，是哈希表的分布式扩展，即在多机分布环境，每台机器承载部分数据的存储情形下，如何通过哈希方式对数据进行增删改查等操作的方法。一致性哈希是DHT的一种实现方法：
§ 一致性哈希可以将海量数据分布到集群中的不同机器节点中，实现数据分片功能。下图是将哈希空间表示为长度为5的二进制数值（m=5）的“一致性哈希”算法的示意图。因为m=5，所以其哈希空间可以表达的数值范围为0～31，“一致性哈希”算法将哈希数值空间按照大小组成一个首尾相接的环状序列。对于每台机器，可以根据其IP和端口号经过哈希函数映射到哈希数值空间内，这样不同的机器就成了环状序列中的不同节点（图1-4中环上的5个大圆即代表不同的机器，分别用Ni来代表，其中的i代表其哈希空间对应的数值），而这台机器则负责存储落在一段有序哈希空间内的数据，比如N14节点就存储主键经过哈希后落在6～14范围内的键值数据，而N5节点则存储30～31以及0～5范围内的键值数据。同时，每个机器节点记录环中的前趋节点和后继节点地址位置，使之成为一个真正的有向环。
§
§ 路由问题：在每个机器节点配置路由表（Finger Table），路由表存储m条路由信息（m为哈希空间的二进制数值比特位长度，上面例子中m=5），其中第i项（0≤i≤m−1）路由信息代表距离当前节点为2i的哈希空间数值所在的机器节点编号。
□
§ 加入新节点：
§ 如果P2P网络中新加入一个机器节点Nnew，首先Nnew必须能够和目前P2P网络中任意一个节点Nx建立联系，通过Nx按照“路由问题”所述路由算法查询Nnew的对应哈希值H（Nnew）=new，可以找到Nnew的后继节点Ns，假设Ns的前趋节点为Np，那么为了能够将Nnew加入P2P网络，需要做以下两件事情：
□ 改变Np、Nnew和Ns对应已经发生变化的前趋节点和后继节点记录，以体现新的网络架构。
□ 其二，数据的重新分片与分布，具体而言就是将Ns节点中存储的应该由Nnew承载的数据（即Ns节点上哈希值小于等于new的记录）迁移到Nnew节点上。
§ 相比于Round Robin数据分片方法，由于其将集群机器数目这一变量从哈希函数中移出，转而将机器及记录主键都映射到哈希数值空间，以此来建立机器与数据记录之间的映射关系。这相当于解除了机器与数据分布函数之间的直接耦合，加入了哈希空间作为解耦层（key-partition和partition-machine映射采用同一个哈希函数），这样不论是新加入机器还是机器故障导致某些机器失效，都只影响当前机器的后继节点上的数据分布，对于集群中其他机器中的数据无任何影响，所以大大增强了数据分片的灵活性。当然，由于其管理P2P网络的复杂性，也导致一致性哈希算法维护成本很高，这是其为提高数据分片灵活性带来的代价。
• 范围分片首先将所有记录的主键进行排序，然后在排好序的主键空间里将记录划分成数据分片，每个数据分片存储有序的主键空间片段内的所有记录。在实现具体存储系统时，往往保持一个数据分片的映射表，记录表每一项记载数据分片的最小主键及其对应的物理机地址（如图）。在对记录／增／删／改时，查找映射表就可以找到对应存储这个记录所在数据分片的物理机，至于数据分片在物理机的管理方式往往采用LSM树（Log Structured Merge Tree），这是一种高效写入的数据索引结构
○
• LSM树：设计思想非常朴素：将对数据的修改增量保持在内存中，达到指定的大小限制后将这些修改操作批量写入磁盘，不过读取的时候稍微麻烦，需要合并磁盘中历史数据和内存中最近修改操作，所以写入性能大大提升，读取效率较低。基于LSM树实现的HBase的写性能比Mysql高了一个数量级，读性能低了一个数量级。
○ LSM树原理把一棵大树拆分成N棵小树，它首先写入内存中，随着小树越来越大，内存中的小树会flush到磁盘中，磁盘中的树定期可以做merge操作，合并成一棵大树，以优化读性能。
○ hbase在实现中，是把整个内存在一定阈值后，flush到disk中，形成一个file，这个file的存储也就是一个小的B+树，因为hbase一般是部署在hdfs上，hdfs不支持对文件的update操作，所以hbase这么整体内存flush，而不是和磁盘中的小树merge update，这个设计也就能讲通了。内存flush到磁盘上的小树，定期也会合并成一个大树。整体上hbase就是用了lsm tree的思路。
第2章 数据复制与一致性
• CAP
○ CAP是对“Consistency/Availability/Partition Tolerance”的一种简称，其分别代表：强一致性、可用性和分区容忍性。对于一个大规模分布式数据系统来说，CAP三要素不可兼得，同一个系统至多只能实现其中的两个，而必须放宽第3个要素来保证其他两个要素被满足。
§ 强一致性：即在分布式系统中的同一数据多副本情形下，对于数据的更新操作体现出的效果与只有单份数据是一样的。
§ 可用性：客户端在任何时刻对大规模数据系统的读／写操作都应该保证在限定延时内完成。
§ 分区容忍性：在大规模分布式数据系统中，网络分区现象，即分区间的机器无法进行网络通信的情况是必然会发生的，所以系统应该能够在这种情况下仍然继续工作。
○ 在绝大多数系统未产生网络分区的情形下，应该尽可能保证AC两者兼得，也即大多数情况下考虑CAP三者兼得，当发生网络分区时，系统应该能够识别这种状况并对其进行正确处理，具体而言，应该分为3个步骤：首先能够识别网络分区发生，然后在网络分区场景下进入明确的分区模式，此时可能会限制某些系统操作，最后在网络分区解决后能够进行善后处理，即恢复数据的一致性或者弥补分区模式中产生的错误。
• ACID原则，ACID是关系数据库系统采纳的原则
○ 原子性（Atomicity）：是指一个事务要么全部执行，要么完全不执行。也就是不允许一个事务只执行了一半就停止。以银行转账为例，这是一个典型的事务，它的操作可以分成几个步骤：首先从A账户取出要转账的金额，A账户扣除相应的金额，之后将其转入B账户的户头，B账户增加相同的金额。这个过程必须完整地执行，否则整个过程将被取消，回退到事务未执行前的状态。不允许出现从A账户已经扣除金额，而没有打入B账户这种情形。
○ 一致性（Consistency）：事务在开始和结束时，应该始终满足一致性约束条件。比如系统要求A+B=100，那么事务如果改变了A的数值，则B的数值也要相应修改来满足这种一致性要求；这里需要注意的是，尽管CAP和ACID都有关于一致性的定义，但是两者的含义是不同的，即两个C代表了不同含义，这点要特别注意。
○ 事务独立（Isolation）：如果有多个事务同时执行，彼此之间不需要知晓对方的存在，而且执行时互不影响，不允许出现两个事务交错、间隔执行部分任务的情形，也即事务之间需要序列化执行。
○ 持久性（Durability）：事务的持久性是指事务运行成功以后，对系统状态的更新是永久的，不会无缘由地回滚撤销。
• 大多数大数据环境下的云存储系统和NoSQL系统则采纳BASE原则
○ 基本可用（Basically Available）。在绝大多数时间内系统处于可用状态，允许偶尔的失败，所以称为基本可用。
○ 软状态或者柔性状态（Soft State），是指数据状态不要求在任意时刻都完全保持同步，到目前为止软状态并无一个统一明晰的定义，但是从概念上是可理解的，即处于有状态（State）和无状态（Stateless）之间的中间状态。
○ 最终一致性（Eventual Consistency）。与强一致性相比，最终一致性是一种弱一致性，尽管软状态不要求任意时刻数据保持一致同步，但是最终一致性要求在给定时间窗口内数据会达到一致状态。
• 一致性模型分类
○
○ 强一致性：对于连接到数据库的所有进程，看到的关于某数据的数据值是一致的，如果某进程对数据进行了更新，所有进程的后续读操作都会以这个更新后的值为基准，直到这个数据被其他进程改变为止。
○ 最终一致性是一种弱一致性。它无法保证某个数值x做出更新后，所有后续针对x的操作能够立即看到新数值，而是需要一个时间片段，在这个时间片段之后可以保证这一点，而在这个时间片段之内，数据也许是不一致的，这个系统无法保证强一致性的时间片段被称为“不一致窗口”（Inconsistency Window）。
○ 因果一致性：发生在进程之间有因果依赖关系的情形下。当进程A将x的数值更新为v2后，会通过Notify(A,B,x,v2）来通知进程B数值已经做出改变，进程B在接收到通知后，之后的操作会以新值作为基础进行读／写，即进程A和进程B保持了数据的因果一致性。而对进程C来说，在不一致窗口内可能还是会看到x的旧数值v1。
○ “读你所写”一致性：是因果一致性的特例：进程A把数据x更新为数值v2后，立即给自己发出了一条通知Notify(A,A,x,v2)，所以进程A之后的操作都是以新数值v2作为基础。其他进程未受影响，在不一致窗口内仍旧可能会看到x的旧数值v1。
○ 单调读一致性：是最终一致性的另外一种变体。它保证如果某个进程读取到数据x的某个版本数据v2，那么系统所有后续的读取操作都不能看到比v2更老版本的数值
○ 单调写一致性：对于某个进程来说，单调写一致性可以保证其多次写操作的序列化，如果没有这种保证，对于应用开发者来说是很难进行程序开发的。
• 副本更新策略：
○ 同时更新：
§ 不通过任何一致性协议直接同时更新多个副本数据。此时存在同时更新的数据不一致问题
§ 通过某种一致性协议预先处理，一致性协议用来唯一确定不同更新操作的执行顺序，这样可以保证数据一致性，但是由于一致性协议有处理成本，所以请求延时会有所增加。
○ 主从式更新：数据的多副本中存在一个主副本（Master Replica），其他副本为从副本，则可被称为主从式更新策略。
§ 同步方式：主副本等待所有从副本更新完成之后才确认更新操作完成，这可以确保数据的强一致性，但是会存在较大请求延时
§ 异步方式：
□ 所有读请求都要通过主副本来响应，即任意一个副本接收到读请求后将其转发给主副本
□ 任意一个副本都可以响应读请求，那么请求延时将会大大降低，但是这可能导致读结果不一致的问题，因为有些副本还存着旧版本的数据信息
§ 混合方式：主副本首先同步更新部分从副本数据，然后即可确认更新操作完成，其他副本通过异步方式获得更新，消息系统Kafka在维护数据副本一致性时即采取此种混合方式。
○ 任意节点更新：数据更新请求可能发给多副本中的任意一个节点，然后由这个节点来负责通知其他副本进行数据更新。所以其特殊性在于：有可能有两个不同客户端在同一时刻对同一个数据发出数据更新请求，而此时有可能有两个不同副本各自响应。
§ 同步通知其他副本
§ 异步通知其他副本
• 一致性协议：
○ 两阶段提交协议（Two-Phrase Commit，2PC）：保证在分布式事务中，要么所有参与进程都提交事务，要么都取消事务，即实现ACID中的原子性（A）的常用手段。
第3章 大数据常用的算法与数据结构
• 布隆过滤器（Bloom Filter，BF）：常被用来检测某个元素是否是巨量数据集合中的成员，具有很好的空间和时间效率，尤其是空间效率极高。BF会产生误判（如果某个成员不在集合中，有可能BF会得出其在集合中的结论），但是不会发生漏判（False Negative）的情况，即如果某个成员确实属于集合，那么BF一定能够给出正确判断。
• SkipList（跳跃表）：是一种可替代平衡树的数据结构，不像平衡树需要强制保持树的平衡，SkipList依靠随机生成数以一定概率来保持数据的平衡分布。
尽管在最坏情况下SkipList的效率要低于平衡树，但是大多数情况下其效率仍然非常高，其插入、删除、查找数据的时间复杂度都是O(log(N))。除了高效外其实现和维护也非常简单，所以在很多大数据系统中在维护有序列表高效读／写的场景下会采用SkipList。就是在插入节点的时候，随机决定该节点应该有多少个指向后续节点的指针，有几个指针就称这个节点是几层的（Level）
• LSM树（Log-structured Merge-tree）的本质是将大量的随机写操作转换成批量的序列写，这样可以极大地提升磁盘数据写入速度，所以LSM树非常适合对写操作效率有高要求的应用场景。但是其对应付出的代价是读效率有所降低，这往往可以引入Bloom Filter或者缓存等优化措施来对读性能进行改善。
• Merkle哈希树：用来在海量数据下快速定位少量变化的数据内容（变化原因可能是损毁、篡改或者正常变化等）。比如在P2P下载系统BitTorrent、Git版本管理工具、比特币以及Dynamo、Riak、Cassandra等NoSQL系统中都得到了应用。
• Snappy是Google开源出的高效数据压缩与解压缩算法库，其目标并非是最高的数据压缩率，而是在合理的压缩率基础上追求尽可能快的压缩和解压缩速度，其压缩和解压缩速度极快，可以在单核处理器上达到250MB/s的压缩效率和500MB/s的解压缩效率。与此同时，Snappy相比其他压缩方案占用CPU时间更少。
• Cuckoo哈希：可以有效解决哈希冲突（Hash Collisions）问题。Cuckoo哈希具有很多优良特性，比如可以在O（1）时间复杂度查找和删除数据，可以在常数时间内插入数据等。其有大约50%的哈希空间利用率。SILT存储系统使用Cuckoo哈希的变体来作为外存数据的索引。SILT的全称是“小索引大表”（Small Index Large Table），这是CMU（美国卡内基·梅隆大学）设计的高效利用内存的高性能基于Flash的KV存储系统，其在Flash存储KV数据，在内存建立数据索引，其存储效率极高，单机可以存储十亿量级的数据并提供很高的读写性能。
第4章 集群资源管理与调度
• 采用独立的资源管理与调度系统而非静态划分资源有如下好处：
○ 集群整体资源利用率高
○ 可增加数据共享能力
○ 支持多类型计算框架和多版本计算框架。
• 资源管理抽象模型：从概念上讲，资源管理与调度系统的主要目的是将集群中的各种资源通过一定策略分配给用户提交到系统里的各种任务，常见的资源主要包括内存、CUP、网络资源与磁盘I/O资源4类。而这个概念模型主要强调三要素：资源组织模型、调度策略和任务组织模型。不同的资源管理与调度系统基本遵循上述概念模型，但是具体在三要素的实现方式上有差异。
○
○ 通用资源管理框架
§
• 调度系统设计的基本问题：
○ 异质性往往指的是组成元素构成的多元性和相互之间较大的差异性。在资源管理与调度场景下，往往有两类异质性需要考虑：资源异质性和工作负载（Workload）异质性
§ 资源异质性：从系统所拥有的资源角度来看的，要考虑这种硬件的资源差异性，一般通过将资源分配单位细粒度划分为较小单元来解决这个问题。
§ 工作负载异质性：因为各种服务和功能特性各异，对资源的需求差异也很大
○ 数据局部性（Data Locality）：为海量数据分布在大规模集群的不同机器中，如果移动数据会产生大量低效的数据网络传输开销，而计算代码相比而言数量小得多，所以移动计算代码到数据所在地而非移动数据到计算任务所在地。
§ 在资源管理与调度语境下，有3种类型的数据局部性：
□ 节点局部性（Node Locality：可以将计算任务分配到数据所在的机器节点，这是数据局部性最优的一种情形，因为完成计算无须任何数据传输）
□ 机架局部性（Rack Locality：虽然计算任务和所需数据分属两个不同的计算节点，但是这两个节点在同一个机架中，这也是效率较高的一种数据局部性，因为机架内机器节点间网络传输速度要明显高于机架间网络传输速度）
□ 全局局部性（Global Locality：需要跨机架进行网络传输，会产生较大的网络传输开销）
○ 抢占式调度与非抢占式调度
§ 抢占式调度是指对于某个计算任务来说，如果空闲资源不足或者出现不同任务共同竞争同一资源，调度系统可以从比当前计算任务优先级低的其他任务中获取已分配资源，而被抢占资源的计算任务则需出让资源停止计算，在后续步骤中继续重新申请新资源来完成后续计算，有时甚至需要废弃已经完成的计算任务重新执行。Omega调度系统采用了抢占式调度方式。
§ 非抢占式调度则只允许从空闲资源中进行分配，如果当前空闲资源不足，则须等待其他任务释放资源后才能继续向前推进。Mesos采用了非抢占式调度。
§ 对于强调高优先级任务执行效率的调度策略来说，往往会采纳抢占式调度方式，以此来保证这些任务即使在资源不足的情况下也能快速完成。而对于更强调资源分配公平性的调度策略来说，往往会采纳非抢占式调度方式。
○ 资源分配粒度（Allocation Granularity）：大数据场景下的计算任务往往由两层结构构成：作业级（Job）和任务级（Task）。一个作业由多个并发的任务构成，任务之间的依赖关系往往形成有向无环图（DAG），典型的MapReduce任务则是一种比较特殊的DAG关系。
§ 一种极端的情况是需要将作业的所有所需资源一次性分配完成，这常被称为“群体分配”（Gang Scheduler）或者“全分或不分”（All-or-Nothing）策略。MPI任务就是一种典型的需要采纳群体分配策略的任务类型。
§ 分配粒度是采取增量满足式分配策略，即对于某个作业来说，只要分配部分资源就能启动一些任务开始运行，随着空闲资源的不断出现，可以逐步增量式分配给作业其他任务以维持作业不断地向后推进，以MapReduce为代表的批处理任务一般采用增量满足式分配策略。
§ 有一种特殊的增量满足式分配策略被称作“资源储备”（Resource Hoarding）策略。这是指只有分配到一定量的资源作业才能启动，但是在未获得足够资源的时候，作业可以先持有目前已分配的资源，并等待其他作业释放资源，这样从调度系统不断获取新资源并进行储备和累积，直到分配到的资源量达到最低标准后开始运行。采取“资源储备”策略的调度，在作业启动前，已分配给该作业的资源一直处于闲置状态。
○ 饿死（Starvation）与死锁（Dead Lock）问题
§ 计算任务“饿死”现象，指的是这个计算任务持续长时间无法获得开始执行所需的最少资源量，导致一直处于等待执行的状态。
§ 死锁问题则是由于资源调度不当导致整个调度系统无法继续正常执行。
§ 调度系统出现死锁必然表现为某些作业处于“饿死”状态，但是有计算任务处于“饿死”情形并不一定意味着调度系统处于死锁状态。
○ 资源隔离方法：资源隔离最常用的手段是Linux容器（Linux Container，LXC），YARN和Mesos都采用了这种方式来实现资源隔离。LXC是一种轻量级的内核虚拟化技术，可以用来进行资源和进程运行的隔离，通过LXC可以在一台物理主机上隔离出多个相互隔离的容器，目前有开源版本。LXC在资源管理方面依赖于Linux内核的cgroups子系统，cgroups子系统是Linux内核提供的一个基于进程组的资源管理的框架，可以为特定的进程组限定可以使用的资源。
• 资源管理与调度范型：
○
○ 集中式调度器（Monolithic Scheduler）：在整个系统中只运行一个全局的中央调度器实例，所有之上的框架或者计算任务的资源请求全部经由中央调度器来满足，因此，整个调度系统缺乏并发性且所有调度逻辑全部由中央调度器来实现。
○ 两级调度器（Two-Level Scheduler）：Mesos、YARN和Hadoop On Demand系统是3个典型的两级调度器系统。
§ 中央调度器：可以看到集群中所有机器的可用资源并管理其状态，它可以按照一定策略将集群中的所有资源分配给各个计算框架，中央调度器级别的资源调度是一种粗粒度的资源调度方式
§ 框架调度器：只能看到由中央调度器分配给自己的资源。各个计算框架在接收到所需资源后，可以根据自身计算任务的特性，使用自身的调度策略来进一步细粒度地分配从中央调度器获得的各种资源。
§ 两级调度器由于在计算框架层面存在第二级资源调度，而这可以提供一种比较天然的并发性，所以整体调度性能较好，也适合大规模集群下的多任务高负载计算情形，具有较好的可扩展性。但是由于中央调度器的存在，使得这种并发是一种悲观并发控制，即中央调度器在做出将某些资源分配给哪个框架的决策过程中，必须依次顺序进行，并需要对目前待决策的资源加锁以避免不同框架的资源申请冲突。而这种悲观并发性会影响系统的整个并发性能。
○ 状态共享调度器（Shared-State Scheduler）：每个计算框架可以看到整个集群中的所有资源，并采用相互竞争的方式去获取自己所需的资源，根据自身特性采取不同的具体资源调度策略，同时系统采用了乐观并发控制手段解决不同框架在资源竞争过程中出现的需求冲突。
• 资源调度策略：
○ FIFO策略：最简单的资源调度策略，提交的作业按照提交时间先后顺序或者根据优先级次序将其放入线性队列相应位置，在资源调度时按照队列先后顺序，先进先出地进行调度与资源分配。FIFO是Hadoop默认的调度策略，很明显这种策略过于简单，在多用户场景下，新加入的作业很容易出现长时间等待调度的现象。
○ 公平调度器（Fair Scheduler）是Facebook为Hadoop开发的多用户多作业调度器。其将用户的任务分配到多个资源池（Pool），每个资源池设定资源分配最低保障和最高上限，管理员也可以指定资源池的优先级，优先级高的资源池会被分配更多的资源，当一个资源池资源有剩余时，可以临时将剩余资源共享给其他资源池。
○ 能力调度器（Capacity Scheduler）是Yahoo为Hadoop开发的多用户调度器，适合用户量众多的应用场景，与公平调度器相比，其更强调资源在用户之间而非作业之间的公平性。
○ 延迟调度策略（Delay Scheduling）往往会作为其他调度策略的辅助措施来增加调度的数据局部性，以此来增加任务执行效率。
○ 主资源公平调度策略（Dominant Resource Fair Scheduling）是Mesos中央调度器采用的公平调度策略，也是最大最小公平算法的一个具体体现。最大最小公平算法的基本思想是：最大化目前分配到最少资源量的用户或者任务的资源量。这个算法常常用来对单个资源进行公平分配，而DRF则将其扩展到了多个资源的公平分配场景下。
• Mesos是美国加州大学伯克利分校AMPLab实验室推出的资源管理与调度系统，从其范型来讲是一个典型的两级调度器。Mesos的设计哲学吸收了类似操作系统中微内核的思想，在中央调度器一级采取极简功能和极小接口，只是根据一定策略决定分配给各个框架多少资源，将数据局部性保证等具体资源调度策略下推到各个框架，这样可以减少中央调度器的负载，增加调度效率，同时也因为其极简设计策略，使得中央调度器支持将来新出现的框架改动最小化，增加了调度系统的可扩展性和健壮性。
• YARN是Hadoop 2.0的重要组成部分，其全称是“另一个资源协调器”（Yet Another Resource Negotiator）。YARN的整体架构图如图4-6所示，其最主要的构件包括：唯一的资源管理器（RM）、每个作业一个的“应用服务器”（AM）以及每个机器一个的“节点管理器”（Node Manager，NM）。RM负责全局的资源管理工作，其内部主要功能部件包括：调度器、AM服务器（AMService/ApplicationMasters，AMS）、Client-RM接口以及RM-NM接口。调度器主要提供各种公平或者能力调度策略，支持可插拔方式，系统管理者可以制定全局的资源分配策略。Client-RM接口负责按照一定协议管理客户提交的作业；RM-NM接口主要和各个机器的NM通过心跳方式进行通信，以此来获知各个机器可用的容器资源以及机器是否产生故障等信息；AMS负责系统内所有AM的最初启动与运行状态管理。
○
○
第5章 分布式协调系统
• Chubby锁服务：由Google公司研发的针对分布式系统协调管理的粗粒度锁服务，一个Chubby实例大约可以负责1万台4核CPU机器相互之间对资源的协同管理。这种锁服务的主要功能是让众多客户端程序进行相互之间的同步，并对系统环境或者资源达成一致认知。
○ Chubby的设计哲学是强调协调系统的可靠性与高可用性及语义易于理解，而不追求处理读／写请求的高吞吐量及在协调系统内存储大量数据。
○ Chubby的理论基础是Paxos一致性协议，Paxos是在完全分布环境下，不同客户端能够通过交互通信并投票，对于某个决定达成一致的算法
○ Chubby与ZooKeeper相比，两者设计思想有重大差异，即ZooKeeper强调系统的高吞吐而Chubby并不追求这一点。
○
○ 从客户端程序来看，Chubby类似于文件系统的目录和文件管理系统，并在此基础上提供针对目录和文件的锁服务。Chubby的文件主要存储一些管理信息或者基础数据，Chubby要求对文件内容一次性地全部读完或者写入，这是为了尽可能地抑制客户端程序写入大量数据到文件中，因为Chubby的目的不是数据存储，而是对资源的同步管理，所以不推荐在文件中保存大量数据。
○ Chubby的会话机制工作如下：客户端向主控服务器发出KeepAlive消息（一个RPC调用），服务器在接收到KeepAlive消息后，阻塞这个RPC调用，直到客户端原先的租约接近过期为止。此时，服务器解除RPC阻塞，KeepAlive调用返回，同时服务器通知客户端说你拥有一个新的租约；客户端在接收到返回信息后立即再次向服务器发出KeepAlive消息，如此循环往复
○ Chubby允许客户端在本地缓存部分服务器数据，而由Chubby来保证缓存数据和服务器端数据完全一致。在很多情况下，客户端所需数据从本地缓存即可读出，这样大大减轻了客户端对服务器的通信压力。为了保持数据一致性，“主控服务器”维护一个缓存表，记录了哪个客户端缓存了什么数据信息；当“主控服务器”接收到某项数据的修改请求时，首先阻塞这个修改数据请求，并查询该缓存表，通知所有缓存该数据的客户端该数据从此无效；客户端在接收到通知后向服务器确认收到该通知，当“主控服务器”接收到所有相关客户端的确认信息后继续执行数据修改请求操作。
• ZooKeeper是Yahoo开发并开源出的一套可扩展高吞吐分布式协调系统，目前已经在各种NoSQL数据库及诸多开源软件中获得广泛使用。正确地使用ZooKeeper可以很方便地解决各种分布式系统的管理协调问题
○ ZooKeeper服务由若干台服务器构成，每台服务器内存中维护相同的类似于文件系统的树形数据结构，其中的一台通过ZAB原子广播协议选举作为主控服务器，其他的作为从属服务器。客户端可以通过TCP协议连接任意一台服务器，如果客户端是读操作请求，则任意一个服务器都可以直接响应请求；如果是更新数据操作（写数据或者更新数据），则只能由主控服务器来协调更新操作；如果客户端连接的是从属服务器，则从属服务器会将更新数据请求转发到主控服务器，由其完成更新操作。
§
○ Zookeeper的内存数据模型类似于传统的文件系统模式，由树形的层级目录结构组成，其中的节点称为Znode，Znode可以使文件，也可以是目录。
§
○ ZooKeeper被广泛使用在各种分布式数据存储与管理系统中，应用开发者可以组合ZooKeeper提供的接口原语来完成各种分布式环境下的协调工作。
§ 领导者选举Leader Election：分布式系统中的主从结构，主控制器负责全局管理控制工作，从节点负责具体的任务计算或数据存储管理。架构清晰、分工明确，但存在主控制器单点的问题。为防止单点失效，往往采取一主一备或一主多备，当主控制器发生故障，由某个台备机接管并成为新的主控机，称为领导者选举。
§ 配置管理Configuration Management：配置文件存储在ZooKeeper某节点Zc中，分布式系统ongoing中的客户端进程在启动时从Zc中读取配置信息，并设置观察标记。若配置沃森将内容在以后被改变，客户端进程会接收到Zc的变化通知，再次读取Zc节点内容以捕获变化点并同时再次设置观察标记，这样以后每次配置文件的变化客户端阿斗可以及时收到通知。
§ 组成员管理Group Membership：动态监控一个组内成员的变化情况。ZooKeeper可以使用临时节点来进行。
§ 任务分配：将不同的任务负载分别分配到多台可用服务器。对于监控进程来说，可以创建任务队列管理节点tasks，所有新进入系统的任务都可以在tasks节点下创建子节点，监控进程观察tasks节点的变化。当有新增任务task-j时ZooKeeper通知监控进程，监控进程找到新增任务并将其分配给机器i，然后在machines目录下对应的m-i节点创建子节点task-j，这意味着将task-j任务分配给了机器m-i。每台工作服务器在machines节点下创建对应子节点，并监听这个子节点的变化，当m-i发现有新增子节点task-j时说明有新分配的任务，可以读出任务信息并执行任务；在执行完task-j后，机器m-i将machines/m-i下的task-j子节点删除，也同时删除tasks节点下的task-j子节点，代表任务已经执行完成，监控进程通过监听tasks可以获知这一情况。通过这种方式可以在监控进程和不同服务器间相互同步来完成任务的分配工作。
§ 锁管理（Locks）：
○ ZooKeeper的实际应用：
§ ZooKeeper在HBase的使用场景包括主控服务器选举与主备切换，作为配置管理在ZooKeeper中存储系统启动信息，发现新的子表服务器及侦测子表服务器是否依然存活等。
§ Twitter的流式计算系统Storm利用ZooKeeper作为主控进程和工作进程状态信息存储场所，使得即使系统出现故障，也可以将进程快速切换到备份机运行。
第6章 分布式通信
• 序列化与远程过程调用框架：
○ 远程过程调用（Remote Procedure Call，RPC）：允许程序调用位于网络中其他机器上的进程，当机器A上的进程调用机器B上的进程时，A上的调用进程被挂起，而B上的被调用进程开始执行，调用方可以通过参数将信息传递给被调用方，然后通过B上的进程返回的结果得到所需的信息。RPC通过以上机制可以隐藏下层的具体通信过程，这大大简化并透明化了网络间进程调用过程，是大规模分布式系统中位于不同机器上进程间通信的黏合剂。RPC框架会融合数据序列化与反序列化功能，以实现高效的数据存取与通信。很多应用直接使用JSON或者XML来作为数据通信的格式。相比较专门的序列化与反序列化框架来说，因为其必须反复传输相同的数据Schema信息，所以在通信效率方面不如专用序列化框架高。
○ 通用的序列化与RPC框架都支持以下特性：接口描述语言（Interface Description Language，IDL）、高性能、数据版本支持以及二进制数据格式。
○ Protocol Buffer（PB）：是在Google内部广泛使用的序列化与RPC框架，是几乎所有Google服务的黏合剂，开源版本被更多地应用于数据序列化方面。与JSON、XML及Thrift等相比，PB对数据的压缩率是最高的。比如ActiveMQ就使用PB作为消息存取工具。
○ Thrift则是Facebook开源出的序列化与RPC框架，在Facebook内部也得到了广泛的使用。因为RPC功能以及IDL语言比PB表达能力更强（Thrift支持List/Set/Map复杂数据结构，PB不支持），所以Thrift的使用场景更丰富，很多开源系统融入Thrift作为RPC构件，比如Hadoop/HBase/Cassandra/Hypertable/ Scribe等。
○ PB和Thrift在使用流程方面大致相同：首先，使用IDL定义消息体以及PRC函数调用接口。其次，使用工具根据上步的IDL定义文件生成指定编程语言的代码。最后，即可在应用程序中链接使用上一步生成的代码。对于RPC功能来说，调用方和被调用方同时引入后即可实现透明网络访问，如果调用方和被调用方采取不同的语言，只要分别根据IDL定义文件生成不同语言库即可实现两者的编码语言解耦。
○ Avro是Apache开源的序列化与RPC框架，使用在Hadoop的数据存储与内部通信中。Avro使用JSON作为IDL定义语言，可以灵活地定义数据Schema及RPC通信协议，提供了简洁快速的二进制数据格式，并能和动态语言进行集成。
• 消息队列：分布式系统构件之间通过传递消息可以解除相互之间的功能耦合，这样可以减轻子系统之间的依赖，使得各个子系统或者构件可以独立演进、维护或者重用。消息队列是在消息传输过程中保存消息的容器或中间件，其主要目的是提供消息路由并保障消息可靠传递。消息是指构件之间信息传递的单位。
○ 消息中间件都支持两种模式的队列
§ 消息队列模式即消息生产者将消息存入队列，消息消费者从队列消费消息
§ Pub-Sub模式则是消息生产者将消息发布到指定主题的队列中，而消息消费者订阅指定主题的队列消息，当订阅的主题有新消息时，消息消费者可以通过拉取（Pull）或者消息中间件通过推送（Push）的方式将消息消费掉
○ Kafka：是Linkedin开源的采用Pub-Sub机制的分布式消息系统，其具有极高的消息吞吐量，较强的可扩展性和高可用性，消息传递低延迟，能够对消息队列进行持久化保存，且支持消息传递的“至少送达一次”语义。
§ 整体架构：主要由3种类型角色构成：消息生产者（Producer）、代理服务器（Broker）和消息消费者（Consumer）。消息生产者产生指定Topic（主题）的消息并将其传入代理服务器集群，代理服务器集群在磁盘存储维护各种Topic的消息队列，订阅了某个Topic的消息消费者从代理服务器集群中拉取（Pull）出新产生的消息并对其进行处理。消费者可以自主控制消费速率，避免采用推送（Push）方式的弊端：如果消费者处理速度跟不上消息生产者产生消息的速度时，消息会大量积压。
□
§ 消息的Topic代表其所属类型，即在内部对应某个名字的消息队列
§ ISR副本管理机制：Kafka通过消息副本机制提供了高可用的消息服务，其副本管理单位不是Topic消息队列，而是Topic的数据分片（Partition）。在配置文件里可以指定数据分片的副本个数，在多个副本里，其中一个作为主副本（Leader），其他作为次级副本（Slave）。所有针对这个数据分片的消息读／写请求都由主副本来负责响应，次级副本只是以主副本数据消费者的方式从主副本同步数据；当主副本发生故障时，Kafka将其中某个次级副本提升为主副本，以此来达到整个消息系统的高可用性。Kafka使用ISR（In-Sync Replicas）机制来保证数据一致性，这个集合备份数据的特点是即时和主副本数据保持一致，而另外一个集合的备份数据允许其消息队列落后于主副本的数据。在做主备切换时，只允许从ISR集合中选择候选主副本，这样即可保证切换后新的主副本数据状态和老的主副本保持一致。在数据分片进行消息写入时，只有ISR集合内所有备份都写成功才能认为这次写入操作成功。在具体实现时，Kafka利用ZooKeeper来保存每个ISR集合的信息，当ISR集合内成员变化时，相关构件也便于通知。
§ Kafka能够高效处理大批量消息的一个重要原因就是将读／写操作尽可能转换为顺序读／写，Kafka涉及将文件内容通过网络进行传输，为了提升效率，Kafka采用了Linux操作系统的SendFile调用。使用SendFile，则可以避免多次数据复制，操作系统可以直接将数据从内核页缓存中复制到网卡缓存，这样可以大大加快整个过程的速度。
□ 正常的网络传输的数据通道：
® 首先，操作系统将数据从磁盘复制到操作系统内核的页缓存中。
® 其次，应用将数据从内核缓存复制到应用空间的缓存中。
® 再次，应用将数据写回内核中的Socket缓存区。
® 最后，操作系统将数据从Socket缓存区复制到网卡的缓存，然后将其通过网络发出。
• 应用层多播通信（Application-Level Multi-Broadcast）：指分布式应用系统内各个节点组织成一定的组织结构，在此结构上实现多接收方的数据通信。
○ Gossip协议：也被称为“感染协议”（Epidemic Protocol），用于研究网络环境下尤其是P2P环境下的多播通信问题
§ 信息传播模型：Gossip协议用来尽快地将本地更新数据通知到网络中的所有其他节点。
□ 全部通知模型（Best Effort或Direct Mail）：当某个节点有更新消息，则立即通知所有其他节点；其他节点在接收到通知后，判断接收到的消息是否比本地消息要新（可以通过时间戳或者版本信息来判断），如果是的话则更新本地数据，否则不采取任何行为。此种信息传播方式简单但是容错性不佳，比如信息发送者如果在通知过程中发生故障抑或消息在通信过程中丢失，都会造成集群中有些节点无法获知最新数据更新内容。
□ 反熵模型（Anti-Entropy）：是最常用的“Gossip协议”，比如Dynamo就用其来进行故障检测。之所以称之为“反熵”，因为我们知道“熵”是信息论里用来衡量系统混乱无序程度的指标，熵越大说明系统越无序、包含的有用信息含量越少；而“反熵”则反其道而行，因为更新的信息经过一定轮数（Round）的传播后，集群内所有节点都会获得全局最新信息，所以系统变得越来越有序，这就是“反熵”的物理含义。
® 在反熵模型中，节点P随机选择集群中另外一个节点Q，然后与Q交换更新信息；Q如果信息有更新，则类似P一样传播给任意其他节点（此时P也可以再传播给其他节点），这样经过一定轮数的信息交换，更新的信息就会快速传播到整个网络节点。其传播过程就是我们常说的“一传十，十传百”的模式。在反熵模型中，P和Q交换信息的方法有以下3种。
◊ Push模式。P将更新信息推送给Q，Q判断是否比本地信息要新
◊ Pull模式。P从Q获取其信息，如果比P本地信息要新，则P更新本地信息
◊ Push-Pull模式。P和Q同时进行Push和Pull操作，即两者同时互相通知对方更新
◊ Push模式与Pull模式相比，其传播效率是刚开始快，但是到后来逐渐变慢，所以整体而言Pull模式传播效率要高于Push模式。实验表明Push-Pull是传播效率最高的，Pull次之，Push相对效率最低
□ 散布谣言模型（Rumor Mongering）：和反熵模型相比，增加了传播停止判断。被拒绝的次数越多越沉默，但它不能保证所有节点都能最终获得更新。更新过程如下：
® 如果节点P更新数据，则随机选择节点Q交换信息；如果节点Q已经被其他节点通知更新了，那么节点P则增加其不再主动通知其他节点的概率，到了一定程度，比如不再通知其他节点的概率达到一定值，则节点P停止通知行为。
§ 应用：Cassandra集群管理
□ Cassandra本质上是使用反熵协议中的Push-Pull模式在节点之间交换最新状态信息。Cassandra集群中有一部分节点被称作“种子节点”（Seeds），这些节点是具有代表性的节点，比如每个数据中心都需要提供至少一个种子节点。新加入的节点进入集群时，首先会和种子节点通信来获得集群的整体状态信息。另外，每个节点都会维护自身的各种状态信息及自己当前看到的集群中其他节点的状态信息，同时状态信息会通过版本号来标明其新旧程度。其中MOVE状态分为BOOT/NORMAL/LEAVING/LEFT这4种状态，标明节点处于启动／正常／准备离开集群／已离开集群4种情形，所以MOVE状态可以用来标明节点自身状态以及其他节点的状态。
第7章 数据通道
• Log数据收集：
○ Log数据收集系统的设计关注点如下
§ 低延迟：从Log数据产生到能够对其分析，希望尽可能快地完成收集过程。当然，因为Log数据的特性，Log数据收集并不要求像流式系统那样的即时性，收集延时在秒级别到分级别甚至小时级别都是可以接受的。
§ 可扩展性：如上所述，Log收集有个特点就是待收集数据的广泛分布性，所以这对Log收集系统的可扩展性有一定要求，因为动态增减服务器及相关服务对于互联网运维来说是常态，Log收集系统应该相应地易扩展、易部署
§ 容错性：同样因为Log收集涉及大量服务器，而这意味着随时有可能发生机器故障，在此约束条件下，如何保证Log收集系统的容错性，不丢失应该收集的数据就是一个必要的要求。
○ Chukwa 是用于针对大规模分布式系统Log收集与分析用途的Apache开源项目，其建立在Hadoop之上。其基本策略是首先收集大量单机的Log增量文件，将其汇总后形成大文件，之后再利用MR任务来对其进行加工处理。
§
§ 工作流程：每台机器节点都部署Chukwa代理程序（Agent），其负责收集应用产生的Log数据并通过HTTP协议传给Chukwa收集器（Collectors）。一般一个收集器负责收集数百个代理程序传来的数据，如果代理程序对应的收集器发生故障，代理程序可以检查收集器列表并从中选择另外一个收集器来发送数据，这样即可实现一定程度的容错。收集器负责将汇总的数据写入HDFS文件中，这些文件被称为DataSink文件。DataSink文件保存的是最原始的Log信息，当其大小达到一定程度，收集器则关闭该文件，随后产生的新数据将被写入新生成的DataSink文件中。ArchiveBuilder进一步合并DataSink文件并做些排序以及去重的工作。Demux是MR程序，负责对原始Log数据进行解析抽取，将原始无结构记录转换为结构化或者半结构化的数据（Chukwa Records）。对于结构化Log数据，可以直接展现给用户，也可以利用MR程序对其进行进一步分析，还可以通过MDL构件将其导入关系数据库中使用SQL语句进行查询。
§ Chukwa整体效率不太高，其主要瓶颈在于Demux，整个数据流动到此后吞吐量急剧下降，这主要是MR任务的启动开销及中间数据和结果数据多次磁盘读／写造成的
○ Scribe：是Facebook开源的分布式日志收集系统，其可以从集群中的机器节点收集汇总Log信息并送达中央数据存储区（HDFS或者NFS），之后可以对其进行进一步的分析处理。Scribe具备高扩展性和高容错能力。
§
• 数据总线：作用就是能够形成数据变化通知通道，当集中存储的数据源（往往是关系型数据库）的数据发生变化时，能尽快通知对数据变化敏感的相关应用或者系统构件，使得它们能尽快捕获这种数据变化。
○ 设计数据总线系统时要关注以下3个特性。
§ 近实时性：因为很多应用希望能尽可能快地捕获数据变化，所以这种变化通知机制越快越好。
§ 数据回溯能力：有时订阅数据变化的应用可能发生故障，导致某一时间段内的数据没有接收成功，此时希望数据总线能够支持数据回溯能力，即应用可以重新获取指定时刻的历史数据变化情况。很明显，支持回溯能力的数据总线可以满足数据的“至少送达一次”（At-Least-Once）语义。
§ 主题订阅能力：因为对于特定的应用来说，其关心的数据是不一样的，将所有数据变化都推送给应用既无必要又浪费系统资源，所以数据总线最好能够支持应用灵活地订阅其关心的数据变化情况
○ 总线系统现实设计中有两个思路：
§ 应用双写：应用将数据变化同时写入数据库以及某个Pub-Sub消息系统中，关注数据变化的应用可以订阅Pub-Sub消息系统中自己关心的主题，以此来获得数据通知，即数据库的归数据库，应用的归消息系统。“应用双写”在实际中可以用在对数据一致性要求不高的场景
§ 数据库日志挖掘：应用先将数据变更写入数据库，数据总线从数据库的日志中挖掘出数据变化信息，然后通知给关心数据变化的各类应用。这样做可以保证数据的一致性，但是实现起来相对复杂，因为需要解析Oracle或者MySQL的日志格式，而且在数据库版本升级后也许旧有的格式作废，需要数据总线也跟着升级。LinkedIn的Databus和Facebook的Wormhole数据总线，两者都是以数据库日志挖掘的思路来实现的。
○ Databus（LinkedIn）：架构如下：
§
§ 为了加快数据通知速度，Databus采用了内存数据中继器（Relay），中继器本质上是个环状的内存缓冲区，之所以设计成环状，是考虑内存大小有限，只能保存一定量的最新更新，所以当更新数据超出缓冲区大小时，相对旧的数据会被新数据覆盖
§ 当数据库发生数据变化的时候，中继器从数据库日志中拉取（Pull）最近提交的事务，并将数据格式转换为较为高效简洁的形式（比如Avro），然后将更新数据存入环状内存缓冲区。客户端侦听中继器数据变化，并将最新的更新数据拉取（Pull）到本地，这里之所以采取拉取而非推送（Push）的方式，是考虑到不同客户端处理数据延时大小不一，拉取的方式更具灵活性，可由客户端自主控制。正因客户端处理数据速度有快有慢，而中继器能够保留的数据相对有限，所以有时客户端会发现所需数据已经被最新的数据覆盖掉，这就是为什么引入Bootstrap的原因，可以将Bootstrap理解为更新数据的长期存储地，而将中继器理解为短期数据存储地。
§ 一般有两种情形客户端会向Bootstrap发出数据请求
□ 一种情况是客户端处理速度慢，所以发现所需数据已经在中继器中被覆盖掉。客户端可以向Bootstrap发出请求，要求获取自从时间T之后的所有更新数据，当客户端逐渐追上中继器数据更新速度后再次改为从中继器获取更新数据。
□ 另外一种情况是新加入系统的新客户端。新加入的客户端首先从Bootstrap获取一份时间T的更新数据快照（Snapshot），然后像第1种情况的客户端一样获取时间T之后的增量更新数据，当客户端逐渐追上中继器的数据更新速度后转向中继器获取之后的更新数据，通过这种方式新客户端就可以获取所有更新数据并像其他客户端一样从中继器获取随后的数据更新了
○ Wormhole（Facebook）：数据库的数据变化可以近实时地通知给同一数据中心或者跨数据中心的缓存系统、Graph Search索引系统以及数据仓库等后台OLAP系统（联机分析处理OLAP（On-Line Analytical Processing），是数据仓库系统的主要应用，支持复杂的分析操作，侧重决策支持，并且提供直观易懂的查询结果）。
§
• 数据导入／导出：最典型的两种数据存储系统之间的迁移是关系型数据库和HDFS之间的数据导入／导出。为了增加这类工作的处理效率，可以考虑使用专用的数据导入／导出系统
○ Sqoop是专门在Hadoop和其他关系型数据库或者NoSQL数据库之间进行相互之间数据导入和导出的开源工具，在其内部实现时，具体的导入／导出工作是通过可以连接并操作数据库的MR任务完成的。
§
第8章 分布式文件系统
• Google文件系统（Google File System，GFS）提供了海量非结构化信息的存储平台，并提供了数据的冗余备份、成千台服务器的自动负载均衡以及失效服务器检测等各种完备的分布式存储功能。
○ GFS设计原则：
§ GFS采用大量商业PC来构建存储集群，稳定性没有很高的保障。因此，数据冗余备份、自动检测机器是否还在有效提供服务、故障机器的自动恢复等都列在GFS的设计目标里。
§ GFS文件系统所存储的文件绝大多数都是大文件，文件大小大多数在100MB到几个GB之间。所以系统的设计应该针对这种大文件的读／写操作做出优化，尽管GFS也支持小文件读／写，但是不作为重点，也不会进行有针对性的操作优化。
§ 系统中存在大量的“追加写”操作，即将新增内容追加到已有文件的末尾，已经写入的内容一般不做更改，很少有文件的“随机写”行为，即指定已有文件中间的某个位置，在这个位置之后写入数据。
§ 对于数据读取操作来说，绝大多数读文件操作都是“顺序读”，少量的操作是“随机读”，即按照数据在文件中的顺序，一次顺序读入较大量的数据，而不是不断地定位到文件指定的位置，读取少量数据。
○ GFS整体架构：
§ GFS文件系统主要由3个组成部分构成：唯一的“主控服务器”（Master，管理工作）、众多的“Chunk服务器”（负责实际的数据存储并响应“GFS客户端”对自己负责的chunk的读／写请求）和“GFS客户端”。
§ GFS在实际存储的时候，首先会将不同大小的文件切割成固定大小的数据块，每一块被称为一个“Chunk”，通常将Chunk的大小设定为64MB，这样，每个文件就是由若干个固定大小的Chunk构成的。GFS即以Chunk为基本存储单位，同一个文件的不同Chunk可能存储在不同的“Chunk服务器”上，每个“Chunk服务器”可以存储很多来自于不同文件的Chunk数据。在“Chunk服务器”内部，会对Chunk进一步切割，将其切割为更小的数据块，每一块被称为一个“Block”，这是文件读取的基本单位，即一次读取至少读一个Block。（如下图所示）。总结来说：GFS命名空间由众多的目录和GFS文件构成，一个GFS文件由众多固定大小的Chunk构成，而每个Chunk又由更小粒度的Block构成，Chunk是GFS中基本的存储单元，而Block是基本的读取单元。
□
§ 下图为GFS整体架构，
□ 主控服务器：主要做管理工作，不仅要维护GFS命名空间，还要维护Chunk的命名空间，每个Chunk都会被赋予一个独一无二的编号，所有Chunk的编号构成了Chunk命名空间，“主控服务器”还记录了每个Chunk存储在哪台“Chunk服务器”上等信息。此外，GFS系统内部就需要维护文件名称到其对应的多个Chunk之间的映射关系。下图为GFS主控服务器所管理的系统数据。维持整个系统正常运转需要3类元数据。
® GFS命名空间和Chunk命名空间：主要用来对目录文件以及Chunk的增删改等信息进行记录。
® 从文件到其所属Chunk之间的映射关系：因为一个文件会被切割成众多Chunk，所以系统需要维护这种映射关系
® 每个Chunk在哪台“Chunk服务器”存储的信息
®
® 在对数据进行备份和迁移的时候，GFS重点要考虑两个因素。一是Chunk数据的可用性。也就是说，如果发现Chunk数据不可用，要及时重新备份。二是要尽可能减少网络传输压力。在不同机器之间传递数据时，因数据量巨大，所以尽可能减少网络传输压力对于系统整体性能表现是很重要的。
® 为了避免“主控服务器”单点失效问题，GFS增加另外一台“影子服务器”（Shadow），当“主控服务器”出现故障无法提供服务时，可由影子服务器接替“主控服务器”行使对应的管理功能。这里需要提及的是，“影子服务器”并非是完整的热备服务器，在“主控服务器”发生故障后，影子服务器可以接替“主控服务器”来使得客户端可以读取元数据，但是其作用仅仅限于此。
□ “GFS客户端”读取数据过程：应用开发者提交的读数据请求是：读取文件file，从某个位置P开始读，读出大小为L的数据。GFS系统在接收到这种请求后，会在内部做转换，因为Chunk大小是固定的，所以从位置P和大小L可以推算出要读的数据位于文件file中的第几个Chunk中，即请求被转换为<文件名file,Chunk序号>的形式。随后，GFS系统将这个请求发送给“主控服务器”，因为“主控服务器”保存了一些管理信息，通过“主控服务器”可以知道要读的数据在哪台“Chunk服务器”上，同时可以将Chunk序号转换为系统内唯一的Chunk编号，并将这两个信息传回到“GFS客户端”。 “GFS客户端”知道了应该去哪台“Chunk服务器”读取数据后，会和“Chunk服务器”建立联系，并发送要读取的Chunk编号以及读取范围，“Chunk服务器”在接收到请求后，将请求数据发送给“GFS客户端”，如此就完成了一次数据读取工作。
□
○ Google的云存储平台有一个显著的特点，就是大量采用“主从结构”，即单一的“主控服务器”和众多的“存储服务器”。采取“主从结构”的好处是：因为整个系统存在一个全局的主控节点，所以管理起来相对简单。相对应的缺点是：因为主控节点是唯一的，很多服务请求都需要经过“主控服务器”，所以很容易成为整个系统的瓶颈，也容易产生单点故障。
○ 系统交互行为：为了方便管理，GFS对于多个相互备份的Chunk，从中选出一个作为“主备份”，其他的被称作“次级备份”，由“主备份”决定“次级备份”的数据写入顺序。
§ 写操作流程：GFS客户端首先和“主控服务器”通信，获知哪些“Chunk服务器”存储了要写入的Chunk，包括“主备份”和两个“次级备份”的地址数据。之后，GFS客户端将要写入的数据推送给3个备份Chunk，备份Chunk首先将这些待写入的数据放在缓存中，然后通知GFS客户端是否接收成功，如果所有的备份都接收数据成功，GFS客户端通知“主备份”可以执行写入操作。“主备份”自己将缓存的数据写入Chunk中，通知“次级备份”按照指定顺序写入数据，“次级备份”写完后答复“主备份”写入成功，“主备份”会通知GFS客户端这次写操作成功完成。如果待写入的数据跨Chunk或者需要多个Chunk才能容纳，则客户端会自动将其分解成多个写操作，其执行流程与上述流程一致。
§
• HDFS：Hadoop中的大规模分布式文件系统。HDFS适合存储大文件并为之提供高吞吐量的顺序读／写访问，不太适合大量随机读的应用场景，也不适合存储大量小文件等应用场景。
○ HDFS的整体架构如下图所示，其由NameNode、DataNode、Secondary NameNode以及客户端构成。NameNode类似于GFS的“主控服务器”，DataNode类似于GFS的“Chunk服务器”，下面分述其各自功能。
§ NameNode：负责管理整个分布式文件系统的元数据，包括文件目录树结构、文件到数据块Block的映射关系、Block副本及其存储位置等各种管理数据。这些数据保持在内存中，同时在磁盘保存两个元数据管理文件：fsimage（内存命名空间元数据在外存的镜像文件）和editlog（各种元数据操作的write-ahead-log文件，在体现到内存数据变化前首先会将操作记入editlog中以防止数据丢失），这两个文件相结合可以构造出完整的内存数据。NameNode还负责DataNode的状态监控，两者通过短时间间隔的心跳来传递管理信息和数据信息。NameNode可以获知每个DataNode保存的Block信息、DataNode的健康状况、命令DataNode启动停止等。如果发现某个DataNode节点发生故障，NameNode会将其负责的Block在其他DataNode机器增加相应备份以维护数据可用性。
§
§ Secondary NameNode：定期从NameNode拉取fsimage和editlog文件并对这两个文件进行合并，形成新的fsimage文件并传回给NameNode。以为了减轻NameNode的工作压力，NameNode本身并不做这种合并操作。所以本质上Secondary NameNode是个提供检查点功能服务的服务器。
§ DataNode类似于GFS的“Chunk服务器”，负责数据块的实际存储和读／写工作，不过HDFS语境下一般将数据块称为Block而非Chunk，Block默认大小为64MB，当客户端上传一个大文件时，HDFS会自动将其切割成固定大小的Block。为了保证数据可用性，每个Block会以多备份的形式存储，默认备份个数为3。
§ HDFS客户端和NameNode联系获取所需读／写文件的元数据，实际的数据读／写都是和DataNode直接通信完成的。其读／写流程和GFS的读／写流程类似，不同点在于HDFS不支持客户端对同一文件的并发写操作，同一时刻只能有一个客户端在文件末尾进行追加写操作。
○ HA方案：Hadoop 2.0给出了一个单点失效的解决方案，“主控服务器”由Active NameNode（简称ANN）和Standby NameNode（简称SNN）一主一从两台服务器构成，ANN是当前响应客户端请求的服务器，SNN作为冷备份或者热备份机，在ANN发生故障时接管客户端请求并由SNN转换为ANN。
§ SNN的所有元数据需要与ANN的元数据保持一致。HA方案通过以下两点保证：
□ 使用第三方共享存储（NAS+NFS）来保存目录文件等命名空间元数据（editlog），ANN将元数据的更改信息写入第三方存储，SNN从第三方存储不断获取更新的元数据并体现在内存元数据中，以此来达到两者的数据一致性。其本质是将NN的单点失效问题转换成为第三方存储的单点失效问题。尽管如此，考虑到很多第三方存储自带很强的冗余与容错机制，所以其可靠性要比单台服务器强得多。
□ 所有DataNode同时将心跳信息发送给ANN和SNN。由于NN中的Block Map信息并不存储在命名空间元数据中，而是在NN启动时从各个DataNode获得的，为了能够使得故障切换时新ANN避免这一耗时行为，所以DataNode同时将信息发送给ANN和SNN。
§ 故障自动切换：HA解决方案采用了独立于NN之外的故障切换控制器（Failover Controller，FC）。FC用于监控NN服务器的硬件、操作系统及NN本身等各种健康状况信息，并不断地向ZooKeeper写入心跳信息，ZooKeeper在此用作“领导者选举”，当ANN发生故障时，ZooKeeper重新选举SNN作为“主控服务器”，FC通知SNN从备份机转换为主控机。在Hadoop系统刚启动时，两台服务器都是SNN，通过ZooKeeper选举使得某台服务器成为ANN。
§ 为了防止在故障切换过程中出现脑裂（Brain-Split）现象，即整个系统中同时有两个或者多个活跃的“主控服务器”，上述HA方案需要在以下3处采取隔离措施（Fencing）。
□ 第三方共享存储：需要保证在任一时刻，只有一个NN能够写入
□ DataNode：需要保证只有一个NN发出与管理数据副本有关的删除命令
□ 客户端：需要保证同一时刻只能有一个NN能够对客户端请求发出正确响应
§
§ 此HA方案有两个明显缺点：其一是第三方存储仍然存在单点失效可能；其二是需要在多处进行隔离措施以防止脑裂现象出现。
○ NameNode联盟：
§ Hadoop 1.x中的HDFS由于采取单一NN的架构，有如下缺点：
□ 命名空间可扩展性差，单一NN情形下，因为所有命名空间数据都需要加载到内存，所以机器物理内存的大小限制了整个HDFS能够容纳文件的最大个数
□ 性能可扩展性差：所有请求都由一台服务器响应，容易达到机器吞吐极限
□ 隔离性差：多租户环境下（Multi-Tenant），单一NN无法在租户之间进行隔离，会造成不可避免的相互影响
§ 从Hadoop 2.0开始，HDFS通过NameNode联盟的方式来解决上述问题。
□ 如果将HDFS的功能高度抽象，可以将其划分为3层：低层是物理存储层，其上是数据块管理层（Block Management），最高层是命名空间管理层。在Hadoop 1.x的架构中，物理存储层是由众多的DataNode承担的，上面两层功能是由单一的NN来承担的。Hadoop2.0的NameNode联盟的核心思想如下图所示，即将一个大的命名空间切割成若干子命名空间，每个子命名空间由单独的NN来负责管理，NN之间独立，相互之间无须做任何协调工作。所有的DataNode被多个NN共享，仍然充当实际数据块的存储场所。而子命名空间和DataNode之间则由数据块管理层作为中介建立映射关系，数据块管理层由若干数据块池（Pool）构成，每个数据块唯一属于某个固定的数据块池，而一个子命名空间可以对应多个数据块池。
®
• HayStack存储系统：是Facebook公司设计开发的一种“对象存储系统”，这里的“对象”主要是指用户上传的图片数据。对象典型的特征是：一次写入，多次读取，从不更改，很少删除，一般将这种数据称为“Blob数据”。设计这种存储系统的时候，保证读取效率是需要重点考虑的要素。
○ 物理卷、逻辑卷
§ 物理卷（Physical Volume）存储多个图片数据对应的某个文件，一般一个“物理卷”文件大小为100GB，可以存储上百万张图片数据
§ 逻辑卷（Logical Volume）不同机器上的若干“物理卷”共同构成一个“逻辑卷”，在HayStack的存储操作过程中，是以“逻辑卷”为单位的，对于一个待存储的图片，会同时将这个图片数据追加到某个“逻辑卷”对应的多个“物理卷”文件末尾
○ Haystack架构：HayStack由3个部分构成：HayStack目录服务、HayStack缓存系统和HayStack存储系统
§ 当Facebook用户访问某个页面时，“目录服务”会为其中的每个图片构造一个URL，通常URL由几个部分构成，典型的URL如下：http://<CDN>/<Cache>/<机器ID>/<逻辑卷ID,图片ID>，
§ <CDN>指出了应该去哪个CDN读取图片，CDN在接收到这个请求后，在内部根据“逻辑卷”ID和图片ID查找图片，如果找到则将图片返回给用户，如果没有找到，则把这个URL的<CDN>部分去掉，将改写后的URL提交给HayStack缓存系统。
§ 缓存系统与CDN功能类似，首先在内部查找图片信息，如果没有找到就会到HayStack存储系统内读取，并将读出的图片放入缓存中，之后将图片数据返回给用户。
§ 这里需要注意的是：“目录服务”可以在构造URL的时候，绕过CDN，直接从缓存系统查找，这样做的目的是减轻CDN的压力。其内部查找过程是一样的。
§
○ 目录服务：采用数据库实现的，它提供多种功能
§ “目录服务”保存了从“逻辑卷”到“物理卷”的映射关系表，这样在用户上传图片和读取图片时可以找到正确的文件。
§ “目录服务”提供了HayStack存储系统的负载均衡功能，保证图片写入和读取在不同机器之间负载是相当的，不至于出现机器之间忙闲不均的状况
§ “目录服务”还决定是将用户请求直接提交给缓存系统还是提交给CDN，以此来对这两者接收到的请求量进行均衡
§ 通过“目录服务”还可以知道哪些“逻辑卷”是只读的，哪些“逻辑卷”可以写入。在有些情况下，某些“逻辑卷”会被标记为只读的，比如其“物理卷”已经基本被写满或者存储系统需要进行调试的时候。
○ HayStack缓存：从功能上讲是与CDN一致的，缓存接收到的访问请求可能来自CDN，也可能直接来自用户浏览器请求。在其内部实现，HayStack采用哈希表的方式存储图片ID和其对应的数据，如果在缓存内没有找到图片，则从HayStack存储系统中读取图片，并加入缓存中，之后将图片内容传给CDN或者直接传递给用户。
○ HayStack存储系统：是整个系统的核心组成部分，对于某台存储机来说，在磁盘存储了若干“物理卷”文件及其对应的索引文件，在内存为每个“物理卷”建立一张映射表，将图片ID存放到“元数据”的映射信息。每个图片的“元数据”包括“删除标记位”、在“物理卷”中的文件起始地址以及图片信息大小，根据文件起始位置和图片大小就可以将图片信息读取出来。对于每个“物理卷”文件，由一个“超级块”和图片数据组成。每个图片的信息被称为一个Needle，每个Needle具体包含图片属性信息，其中比较重要的属性信息包括图片唯一标记Key和辅助Key、删除标记位、图片大小以及图片数据，除此之外还包含一些管理属性以及数据校验属性。
§
○ 读取图片的请求来说，HayStack缓存系统会向存储系统提供图片的“逻辑卷”ID编号以及图片ID（由Key和辅助Key构成），当存储系统接收到请求后，会在内存中的“物理卷”映射表中查找图片ID，如果找到，则根据映射表保存的信息可以获取其在对应“物理卷”中的文件起始位置和文件大小，如此就可以读到这个图片的内容。
○ 上传图片请求，HayStack存储系统根据Web服务器传过来的图片“逻辑卷”ID编号以及图片ID和图片数据，将这个图片信息追加到对应的“物理卷”文件末尾，同时在内存的映射表中增加相应的映射信息。
○ 如果用户更改了图片的内容后再次上传，HayStack存储系统不允许覆盖原先图片信息这种操作，因为这种操作严重影响系统效率，而是将这个修改的图片当作一个新的图片追加到“物理卷”的文件末尾，不过这个图片的ID是不变的。此时有两种情况：一种情况是更改后的图片的“逻辑卷”ID和原始图片的“逻辑卷”ID不同，这样更新图片会写入到不同的“物理卷”中，此时“目录服务”修改图片ID对应的逻辑卷映射信息，此后对这个图片的请求就直接转换到更新后的图片，原始图片不会再次被访问；另外一种情况是更改后图片的“逻辑卷”与原始图片的“逻辑卷”相同，此时HayStack存储系统将新图片追加到对应的“物理卷”末尾，也就是说，同一个“物理卷”会包含图片的新旧两个版本的数据，但是由于“物理卷”是顺序追加的，所以更改后的图片在“物理卷”中的文件起始位置一定大于原始图片的起始位置，HayStack在接收之后的用户请求时会进行判断，读取文件起始位置较大的那张图片信息，这样就保证读取到最新的图片内容。
○ 如果用户删除某张图片，HayStack系统的操作也很直观，只要在内存映射表和“物理卷”中在相应的“删除标记位置”上做出标记即可。系统会在适当的时机回收这些被删除的图片数据空间。
• 文件存储布局：GFS或者HDFS文件中记录的存储布局方式对于这些构建在上面的分析与存储系统来说非常重要，不同的存储布局方案会对上层系统的整体性能有着重要、甚至起决定性作用的影响。具体而言，底层文件存储布局对于将数据加载入数据仓库的效率、响应用户查询的速度，以及对于底层存储架构磁盘空间利用率的提升都有直接且重要的影响。常见的三种文件存储布局如下：
○ 行式存储：广泛使用在主流关系型数据库及HDFS文件系统中，每条记录的各个字段连续存储在一起，而对于文件中的各个记录也是连续存储在数据块中的。下图是HDFS的行式存储布局，每个数据块除了存储一些管理元数据外，每条记录都以行的方式进行数据压缩后连续存储在一起。
§
§ 大数据分析系统来说，行式存储布局有两个明显缺陷：其一是对于很多SQL查询来说，其所需读取的记录可能只涉及整个记录所有字段中的部分字段，而若是行式存储布局，即使如此也要将整个记录全部读出后才能读取到所需的字段；其二是尽管在存储时可以使用数据压缩模式，但是对于记录的所有字段只能统一采用同一种压缩算法，这样的压缩模式导致数据压缩率不高，所以磁盘利用率不是很高。
§ 优势：如果应用需要按行遍历或者查找数据，此时较适合使用此种存储布局，因为行数据连续存储，所以能够一次性地将所有字段的内容读出，而且同一记录的内容一定在一个数据块中，不像列式存储布局一样，为了读取完整记录内容，可能需要一些跨网络的数据读取操作。
○ 列式存储：在实际存储数据时，按照列对所有记录进行垂直划分，将同一列的内容连续存放在一起。简单的记录数据格式类似于传统数据库的（记录-字段）这种平面型（Flat）数据结构，一般的列式存储布局采取列族（Column Group/Column Family）的方式；而复杂的记录格式则可能是嵌套（Nested）的记录结构，即字段内容可能是另外一个有结构的记录体，比如JSON格式就是这种支持嵌套表达的复杂记录格式，Google在海量数据交互式分析系统Dremel中提出了一种列式存储布局方案。
§ 列族方式：典型的列式存储布局是按照记录的不同列，对数据表进行垂直划分，同一列的所有数据连续存储在一起。
□ 两个好处：
® 对于上层的大数据分析系统来说，如果SQL查询只涉及记录的个别列，则只需读取对应的列内容即可，增加I/O效率
® 因为数据按列存储，可以针对每列数据的类型采取具有针对性的数据压缩算法，不同的字段可以采用不同的压缩算法，这样整体压缩效果会有极大的提升
□ 缺点：当需要遍历每条数据记录，并处理记录的各个字段或者多个字段，列式存储要从列式数据中拼合出原始记录内容，这样对于HDFS这种按块存储的模式，有可能不同列的内容分布在不同数据块，而不同的数据块在不同的机器节点上，所以为了拼合出完整记录内容，可能需要大量的网络传输才行，很明显这样效率会比较低下
□ 列族：将记录的列进行分组，将经常一起使用的列分为一组，这样即使是按照列式来存储数据的，也可以将经常联合使用的列数据存储在一个数据块中，避免不必要的网络传输来获取多列数据。BigTable和HBase的底层GFS或者HDFS存储布局采用列族方式
§ Dremel的列存储方式：Google开发的针对海量数据进行交互式查询与数据分析应用场景的大数据分析系统，可以对存储在数以千计GFS或者BigTable服务器上的海量数据进行秒级别的SQL查询，其中很关键的一点就是针对复杂嵌套数据的列式存储。
○ 混合式存储：融合了行式和列式存储各自的优点，首先其将记录表按照行进行分组，若干行划分为一组，而对于每组内的所有记录，在实际存储时按照列将同一列内容连续存储在一起。一方面可以像行式存储一样，保证同一行的记录字段一定是在同一台机器节点上的，避免拼合记录的网络传输问题；另外一方面可以像列式存储布局一样按列存储，这样不同列可以采用不同的压缩算法，同时也可以避免读取无关列的数据。
§ 典型的混合式存储方案包括RCFile、ORCFile和Parquet。其中，RCFile和ORCFile已经集成进入Hive系统，而Parquet则是Twitter模仿Dremel的列式存储开发并开源出的文件布局方案
§ RCFile：典型的混合式存储布局，将记录表内的记录按照行划分为行组（Row Group），HDFS每个数据块可以包含多个行组数据。对于每个行组，存储3类信息：
□ Sync是行组同步标识，用于识别是否是数据块中一个新的行组开始；
□ 元数据（Metadata Header）则记录了这个行组包含多少记录，每列占用空间大小等数据
□ 实际数据，在数据块中按照列式存储
□
§ ORCFile：针对RCFile提出的优化的文件存储布局方案。ORCFfile包含若干数据行组，每个数据行组被称为数据带（Stripe），文件尾（File Footer）记录文件中所有数据带的元信息，比如有多少个数据带，每个数据带包含的记录个数及每列采用何种数据压缩算法等信息，同时也记录每列的统计信息，比如该列的最大值、最小值等。附录（Postscript）中记载了压缩算法的参数信息。每个数据带由3类信息构成：行数据区（Row Data）按列存储该行组记录的实际数据；数据带尾（Stripe Footer）记录压缩数据流的位置信息；索引数据（Index Data）记录了该行组所有记录中每一列的最大值和最小值，另外还记录了行组内部分记录的每一列字段在行数据区的位置信息，即记录的索引信息。利用这些记录索引信息，可以在查找记录时跳过不满足条件的记录，提高执行效率。
• 纠删码（Erasure Code）：在不对数据做备份的情形下提供类似的数据可靠性
○ 一种常见的做法是：对于热点数据，在大规模存储系统中仍然保留3个备份，而对于冷数据，则只保留1份数据，通过纠删码来保证数据的可靠性。之所以不对所有数据都采用纠删码的方式，是因为备份数据除了能够增加数据的可用性外，还可以提升数据的并发读取效率，所以对于热点数据用多备份的方式比较合适。
○ 纠删码通过对原始数据进行校验并保留校验数据，以增加冗余的方式来保证数据的可恢复性。极大距离可分码（Maximum Distance Separable codes，MDS）是一种非常常用的纠删码，其将数据文件切割为等长的n个数据块，并根据这n个数据生成m个冗余的校验信息，这样使得n+m块数据中即使任意m块数据损失，也可以通过剩下的n块数据对m块损失的数据进行重构，以此来完成数据容错功能。对于上述参数配置，一般称为满足（n,m)配置的MDS。
○ Reed-Solomon（简称RS）编码和局部可修复编码（Locally Repairable Codes，LRC）是两种经典的纠错码。其中RS编码是最典型的MDS编码，目前在Google的Colossus以及Facebook的HDFS-RAID都已经引入这种编码。LRC并非MDS编码，这是针对RS编码在分布式存储中面临的问题而在最近提出的一种编码，微软的AWS云存储系统及Facebook的Xorbas系统都采用了这种编码。
第9章 内存KV数据库
• RAMCloud：是斯坦福大学提出的大规模集群下的纯内存KV数据库系统，最大的特点是读／写效率高，其设计目标是在数千台服务器规模下读取小对象速度能够达到5～10纳秒，这种速度是目前常规数据中心存储方案性能的50～1000倍
○ RAMCloud的整体架构如下：
§
§ 存储服务器由高速网络连接，每台存储服务器包含两个构件：Master和Backup。Master负责内存KV数据的存储并响应客户端读／写请求，Backup负责在外存存储管理其他服务器节点内存数据的数据备份。
§ 每个RAMCloud集群内包含唯一的管理节点，称之为协调器（Coordinator）。协调器记载集群中的一些配置信息，比如各个存储服务器的IP地址等，另外还负责维护存储对象和存储服务器的映射关系，即某个存储对象是放在哪台服务器的。RAMCloud的存储管理单位是子表（Tablet），即若干个主键有序的存储对象构成的集合，所以协调器记载的其实是子表和存储服务器之间的映射关系。
§ 为了增加读／写效率，客户端在本地缓存一份子表和存储服务器的映射表，当有对应数据读／写请求时，直接从缓存获取记录主键所在的存储服务器地址，然后直接和存储服务器进行交互即可，这样也能有效地减轻协调器的负载。但是这会导致以下问题：当子表被协调器迁移后，客户端的缓存映射表会过期。RAMCloud的解决方案为：当客户端发现读取的记录不在某台存储服务器时，说明本地缓存过期，此时可以从协调器重新同步一份最新的映射表，之后可以重新对数据进行操作。
○ 为了能够支持快速数据持久化以及故障时快速数据恢复，RAMCloud在内存和外存存储数据时都统一采用了LSM树方案，其对应的Log结构被切割为8MB大小的数据片段（Segment）。RAMCloud的副本管理策略如下
§
§ 当RAMCloud接收到写数据请求时，首先将其追加进入内存中的Log结构中，然后更新哈希表以记载记录在内存中的存储位置，这里之所以会需要哈希表，是因为内存数据采取LSM树结构后，是由若干个Log片段构成的，所以需要记载记录所在Log片段的位置信息。
§ RAMCloud的主数据服务器将新数据转发给其他备份服务器，备份服务器将新数据追加到内存中Log片段后即通知主数据服务器返回，主数据服务器此时即可通知客户端写操作成功。因为整个备份过程都是内存操作不涉及外存读／写，所以这样做速度较快。当备份服务器用于备份的Log片段写满时将其写入外存的LSM结构中。
§ 当某台存储服务器发生故障时，其负责的对应数据此时不可用，RAMCloud的策略是尽可能快速地从备份数据中重建内存数据。
• Redis：不仅支持基本数据类型，也支持列表、集合等复杂数据结构，所以有较强的表达能力，同时有非常高的单机读／写效率。Redis在发展过程中更强调单机系统读／写性能和系统的使用便捷性，在高可用性方面做得一直不太理想
○ Redis的数据副本维护策略：
§
§ 当Slave初次启动时，从Master获取数据，在数据复制过程中，Master是非阻塞的，即同时可以支持读／写操作。Master采用快照加增量的异步方式完成数据复制过程，首先在时刻T将内存数据写入本地快照文件，同时在内存记录从T时刻起新增的数据操作，当快照文件生成结束后，Master将文件传给Slave，Slave先保存为本地文件，然后将其加载入内存。之后，Master将T时刻后的数据变更操作以命令流的形式传给Slave，Slave顺序执行命令流，这样就达到数据和Master保持同步。
§ 如果Master和Slave之间的连接因某种原因中断，在2.8版之前，Slave再次和Master建立连接后需要完全重新复制一遍数据，2.8版本对此进行了改进，支持增量更新。Master在内存维护命令流记录，同时，Master和Slave都记载上次复制时的命令流地址（Offset），当Slave重新连接Master时，Master可以根据地址偏移量将增量更新传递给Slave。
§ 由于Redis的主从复制采用异步方式，所以Master接收到数据更新操作与Slave接收到数据副本有一个时间差，这样如果Master发生故障可能会导致数据丢失。另外，因为Redis并未支持主从自动切换，如果Master故障，很明显此时系统对外表现为只读不能写入。
○ 在实际使用Redis时，很多场景对数据高可用性有较高要求，一种常见的解决思路是使用Keepalived结合虚拟IP来实现Redis的HA方案。Keepalived是软件路由系统，主要的目的是为应用系统提供简洁强壮的负载均衡方案和通用高可用方案。使用Keepalived实现Redis高可用方案思路如下
§ 在两台服务器（或者多台，机制类似）分别安装Redis并设置成一主一备
§ Keepalived配置虚拟IP和两台Redis服务器IP的映射关系，这样，对外统一采用虚拟IP，而虚拟IP和真实IP的映射关系及故障切换由Keepalived来负责。当Redis服务器都正常时，数据请求由Master负责，Salve只需从Master同步数据；当Master发生故障时，Slave接管数据请求，同时关闭主从复制功能以避免Master再次启动后Slave数据被清掉；当发生故障的Master恢复正常后，首先从Slave同步数据以获得最新的数据情况，然后关闭主从复制功能并恢复Master身份，与此同时Slave恢复其Slave身份
• MemBase：集群环境下的内存KV数据库，目前已更名为CouchBase。MemBase缘起于Zynga的实际需求：在社交游戏环境下，需要高速、可靠且支持高吞吐量的存储系统，尤其是对写操作的效率要求很高。
○ MemBase的整体架构
§
§ MemBase中的所有服务器都是地位平等的，但是其数据副本管理采用了Master-Slave模式。每个虚拟桶有一台服务器作为主数据存储地，这台服务器负责响应客户端请求，副本存放在其他服务器内存中，其副本个数可以通过配置来指定。每个虚拟桶有一台服务器作为主数据存储地，这台服务器负责响应客户端请求，副本存放在其他服务器内存中，其副本个数可以通过配置来指定。
○ MemBase作为内存KV数据库，从架构设计上有比较完善的系统高可用性保障措施，但是就像本章开头所述，这种方式的缺点是所有副本数据放在内存，所以存储成本较高。
第10章 列式数据库
• 列式数据库兼具NoSQL数据库和传统数据库的一些优点，其具备NoSQL数据库很强的水平扩展能力、极强的容错性以及极高的数据承载能力，同时也有接近于传统关系型数据库的数据模型，在数据表达能力上强于简单的Key-Value数据库。从列式数据库的技术发展趋势可以看出，其发展方向是越来越多地融合和兼具两者的优点，包括全球范围的数据部署、千亿级别的数据规模、极低的数据读／写延迟、类SQL操作接口、分布式事务支持等。
• BigTable：是一种针对海量结构化或者半结构化数据的存储模型，在Google的云存储体系中处于核心地位。
○ BigTable的数据模型：BigTable本质上是一个三维的映射表，其最基础的存储单元是由（行主键、列主键、时间）三维主键（Key）所定位的。
§
§ BigTable中的列主键包含两级，其中第一级被称为“列家族”（Column Families），第二级被称为“列描述符”（Qualifier），两者共同组成一个列的主键，列主键=“列家族：列描述符”
§ BigTable内可以保留同一信息随着时间变化的不同版本，这个不同版本由“时间”维度来进行表达。
§ BigTable的数据模型粗看很像关系型数据库的关系模型，但是两者有很大差别，关系型数据库的列在设计表格之初就已经指定，而BigTable是可以随时对表格的列进行增删的，而且每行只存储列内容不为空的数据，这被称作“模式自由型”（Schema Free）数据库。
○ BigTable整体结构：包含主控服务器（Master Server）、子表服务器（Tablet Server）和客户端程序（Client）
§ 每个表格将若干连续的行数据划分为一个子表（Tablet），这样，表格的数据就会被分解为一些子表。
§ “子表服务器”主要负责子表的数据存储和管理，同时需要响应客户端程序的读／写请求，其负责管理的子表以GFS文件的形式存在，BigTable内部将这种文件称之为SSTable，一个子表就是由“子表服务器”磁盘中存储的若干个SSTable文件组成的。
§ “主控服务器”负责整个系统的管理工作，包括子表的分配、子表服务器的负载均衡、子表服务器失效检测等。
§ “客户端程序”则是具体应用的接口程序，直接和“子表服务器”进行交互通信，来读／写某个子表对应的数据。
§
○ BigTable的管理数据：
§ BigTable利用Chubby系统和一个被称为“元数据表”（MetaData Table）的特殊表格来共同维护系统管理数据。“元数据表”是BigTable中一个起着特殊作用的表，这个表格的每一行记载了整个BigTable中某个具体子表存储在哪台“子表服务器”上等管理信息，但是它一样也会被切割成若干子表并存储在不同的“子表服务器”中。表的第1个子表被称为“Root子表”，用来记录“元数据表”自身除“Root子表”外其他子表的位置信息，因为“元数据表”的子表也是分布在不同机器上的，所以通过“Root子表”的记录就可以找到“元数据表”中其他子表存储在哪台机器上，即通过“Root子表”可以找到完整的“元数据表”。“元数据表”中其他子表的每一行，则记录了BigTable中应用程序生成的表格（用户表）的某个子表的管理数据。其中，每一行以“用户表表名”和在这个子表内存储的最后一个“行主键”共同构成“元数据表”内此条记录的“行主键”，在记录行的数据里则存储了这个子表对应的“子表服务器”等其他管理信息。
□
○ 主控服务器（Master Server）：在BigTable中专门负责管理工作
○ 子表服务器（Tablet Server）是BigTable系统中用来存储和管理子表数据的，从具体功能来讲，子表服务器支持以下功能。
§ 存储管理子表数据，包括子表存储、子表恢复、子表分裂、子表合并等。
§ 响应客户端程序对子表的写请求
§ 响应客户端程序对子表的读请求
○ Apache HBase，可以将其看作BigTable的开源版本，目前已广泛应用在各种大数据实时存取场景中
• PNUTS存储系统：Yahoo公司构建的提供在线数据服务的列式云存储系统。PNUTS采取了弱一致性模型，以这种宽松的一致性模型为代价，换取系统更好的可扩展性、高可用性以及强容错性。
○ PNUTS在以下几方面有其特色。
§ 支持在线实时请求的响应。
§ 支持多数据中心的分布式存储和数据备份与同步。
§ 很多云存储系统对于数据更新，采取先写入系统Log文件，事后回放（Replay）的方式来保证数据操作的容错性。PNUTS则采取了“消息代理”的机制来保证这一点，虽然从本质上说也是类似于Log回放机制，但是其表现形式并不相同。
§ 数据的一致性，PNUTS采取了以记录为单位的“时间轴一致性”。
○ PNUTS的整体架构：
§ PNUTS支持数据的多数据中心部署，每个数据中心被称为一个“区域”（Region），每个区域所部署的系统都是完全相同的，每条记录在每个区域都有相应的备份。每个区域内主要包含3个基本单元：“子表控制器”、“数据路由器”和“存储单元”，其中“存储单元”负责实际数据的存储，其他两个部分起到数据管理的作用，“消息代理”则横跨多个区域，主要负责数据在不同区域的更新与同步。
§
• MegaStore：目前大多数互联网应用中相当重要的一部分需要与用户进行实时交互，如何针对这些应用的特点构造海量存储系统？这是非常具有挑战性的问题。MegaStore即是Google针对这类应用自行研发的海量存储系统。
○ MegaStore的基本思路是：将大规模数据进行细粒度切割，切分成若干实体群组（Entity Group），在实体群组内提供满足ACID语义的强一致性服务，但是在实体群组之间提供相对弱些的数据一致性保证。利用改造的Paxos协议来将数据分布到多个数据中心，这样同时满足了数据请求的高速度低延迟以及高可用性，可用性是通过将数据分布到不同数据中心获得的，而数据请求的高速度低延迟则是靠优化后的Paxos协议来保证的。
• Spanner：是Google开发的可在全球范围部署的具有极强可扩展性的列式数据库系统，其可以将千亿规模的数据自动部署到世界范围数百个数据中心中的百万台服务器中，通过细粒度的数据备份机制极大地提高了数据的可用性以及地理分布上的数据局部性。Spanner具备数据中心级别的容灾能力，即使整个数据中心完全遭到破坏也可以保证数据的可用性。除此之外，Spanner还具备接近于传统数据库关系型模型的半结构化数据模型定义、类SQL查询语言以及完善的事务支持等特性。
第11章 大规模批处理系统
• MapReduce：由Google公司于2004年提出的，这不仅仅是一种分布式计算模型，同时也是一整套构建在大规模普通商业PC（成千台机器）之上的批处理计算框架，这个计算框架可以处理以PB计的数据，并提供了简易应用接口，将系统容错以及任务调度等设计分布式计算系统时需考虑的复杂实现很好地封装在内，使得应用开发者只需关注业务逻辑本身即可轻松完成相关任务。
○ 计算模型：对于某个计算任务来说，其输入是Key/Value数据对，输出也以Key/Value数据对方式表示
§ Map函数以Key/Value数据对作为输入，将输入数据经过业务逻辑计算产生若干仍旧以Key/Value形式表达的中间数据。MapReduce计算框架会自动将中间结果中具有相同Key值的记录聚合在一起，并将数据传送给Reduce函数内定义好的处理逻辑作为其输入值。
§ Reduce函数接收到Map阶段传过来的某个Key值及其对应的若干Value值等中间数据，函数逻辑对这个Key对应的Value内容进行处理，一般是对其进行累加、过滤、转换等操作，生成Key/Value形式的结果，这就是最终的业务计算结果。
○ Google的MapReduce计算框架架构：当用户程序执行MapReduce提供的调用函数时，
§ 处理流程如下
□ MapReduce框架将应用的输入数据切分成M个数据块，典型的数据块大小为64MB，然后可以启动位于集群中不同机器上的若干程序
□ 程序中有一个全局唯一的主控Master程序以及若干工作程序（Worker），Master负责为Worker分配具体的Map任务或者Reduce任务并做一些全局管理功能。Master将任务分配给目前处于空闲状态的Worker程序。
□ 被分配到Map任务的Worker读取对应的数据块内容，从数据块中解析出一个个Key/Value记录数据并将其传给用户自定义的Map函数，Map函数输出的中间结果Key/Value数据在内存中进行缓存。
□ 缓存的Map函数产生的中间结果周期性地被写入本地磁盘，每个Map函数的中间结果在写入磁盘前被分割函数（Partitioner）切割成R份，R是Reduce的个数。Map函数完成对应数据块的处理后将其R个临时文件位置通知Master，再由Master将其转交给Reduce任务的Worker。
□ 当某个Reduce任务Worker接收到Master的通知时，其通过RPC远程调用将Map任务产生的M份属于自己的数据文件远程拉取（Pull）到本地。只有所有Map任务都完成时Reduce任务才能启动。当所有中间数据都拉取成功，Reduce任务根据中间数据的Key对所有记录进行排序，这样就可以将具有相同Key的记录顺序聚合在一起。这里需要强调的是：Reduce任务从Map任务获取中间数据时采用拉取方式而非由Map任务将中间数据推送（Push）给Reduce任务，这样做的好处是可以支持细粒度容错。假设在计算过程中某个Reduce任务失效，那么对于Pull方式来说，只需要重新运行这个Reduce任务即可，无须重新执行全部所有的Map任务。而如果是Push方式，这种情形下只有所有Map任务都全部重新执行才行。因为Push是接收方被动接收数据的过程，而Pull则是接收方主动接收数据的过程。
□ Reduce任务Worker遍历已经按照中间结果Key有序的数据，将同一个Key及其对应的多个Value传递给用户定义的Reduce函数，Reduce函数执行业务逻辑后将结果追加到这个Reduce任务对应的结果文件末尾
□ 当所有Map和Reduce任务都成功执行完成时，Master便唤醒用户的应用程序，此时，MapReduce调用结束，进入用户代码执行空间
□
§ Google的MapReduce框架支持细粒度的容错机制。Master周期性地Ping各个Worker，如果在一定时间内Worker没有响应，则可以认为其已经发生故障。此时将由这个Worker已经完成的和正在进行的所有Map任务重新设置为Idle状态，这些任务将由其他Worker重新执行。但因为Master是单点的，所以如果Master失败，则整个MapReduce任务失败，应用可以通过反复提交来完成任务。
○ MapReduce运算机制的优势是数据的高吞吐量、支持海量数据处理的大规模并行处理、细粒度的容错，但是并不适合对时效性要求较高的应用场景，比如交互式查询或者流式计算，也不适合迭代运算类的机器学习及数据挖掘类应用，主要原因有以下两点：
§ 其Map和Reduce任务启动时间较长。
§ 在一次应用任务执行过程中，MapReduce计算模型存在多处的磁盘读／写及网络传输过程。
• MapReduce计算模式：
○ 求和模式（Summarization Pattern）：
§ 数值求和：计算对象是数值类型，统计计算包括简单计数、求最小值／最大值、求平均值／中位数等
§ 记录求和：对于非数值的情况，往往需要将非数值内容进行累加形成队列，一般应用中累加内容是对象的ID
○ 过滤模式（Filtering Pattern）：
§ 简单过滤：即根据一定条件从海量数据中筛选出满足条件的记录。
§ TOP10
○ 组织数据模式（Data Organization Pattern）
§ 数据分片
§ 全局排序
○ Join模式（Join Pattern）两个数据集合进行Join操作也较常见，就是将两个不同数据集合内容根据相同外键进行信息融合的过程
§ Reduce-Side Join：具有实现简单以及具备通用性的优点，但是缺点是因为其没有根据不同Join类型的特点做出特定优化，所以计算效率较低
§ Map-Side Join：两个需要Join的数据集合L和R，一个大一个小（假设L大R小），而且小的数据集合完全可以在内存中放入，此时，只需要采用一个Map-Only MapReduce任务即可完成Join操作。Mapper的输入数据块是L进行拆分后的内容，而由于R足够小，所以将其分发给每个Mapper并在初始化时将其加载到内存存储，一般比较高效的方法是将R存入内存哈希表中，以外键作为哈希表的Key，这样即可依次读入L的记录并查找哈希表来进行Join操作。Map-Side Join处理效率要高很多，但是其必须满足R小到可以在内存存储这一前提条件。
• DAG计算模型：DAG是有向无环图（Directed Acyclic Graph），在大数据处理领域，DAG计算模型往往是指将计算任务在内部分解成为若干个子任务，这些子任务之间由逻辑关系或运行先后顺序等因素被构建成有向无环图结构。
○ DAG计算系统的三层结构
§ 最上层是应用表达层，即通过一定手段将计算任务分解成由若干子任务形成的DAG结构，这层的核心是表达的便捷性，主要目的是方便应用开发者快速描述或者构建应用。
§ 最下层是物理机集群，即由大量物理机器搭建的分布式计算环境，这是计算任务最终执行的场所
§ 中间层是DAG执行引擎层，其主要目的是将上层表达的DAG计算任务通过转换和映射，部署到下层的物理机集群中运行。中间层是DAG计算系统的核心，计算任务的调度、底层硬件的容错、数据与管理信息的传递、整个系统的管理与正常运转等都需要由这层来完成。
○ Dryad：微软的批处理DAG计算系统，其主要目的是为了便于开发者便捷地进行分布式任务处理，其是在大数据处理领域较早明确提出DAG计算模型的系统。
§ Dryad架构：
□ 任务管理器（Job Manager，JM）负责将逻辑形式存在的DAG描述映射到物理集群机器中，起到任务调度器以及全局管理者的作用。
□ 命名服务器（Name Server，NS）负责维护集群中当前可用的机器资源，从命名服务器还可以查到机器在物理拓扑结构中的位置，这样便于JM在任务调度时考虑到数据局部性，尽可能地将计算推送到数据节点上，以此来减少网络通信开销，加快任务执行速度。
□ 集群中每台工作机上安装Daemon守护进程（图中标为D的即是），其作为JM在计算节点上的代理，负责具体子任务的执行和监控。
□ 当DAG中某个图节点首次被分配到工作机上时，JM将二进制代码传输给Daemon进程由其来负责执行，Daemon进程在执行过程中和JM通信，来汇报执行进度以及数据处理完成情况。DAG图节点之间的有向边代表数据流动的方向，分配到DAG图节点的工作机之间，直接进行通信来传输数据，并不需要JM介入，所以JM不会成为整个系统的瓶颈。
□
○ FlumeJava是Google内部开发的DAG系统，其本质上是在MapReduce基础上的DAG计算系统，图中每个节点可以看作是单个的MapReduce子任务，FlumeJava还提供了常用的操作符来建立图之间的边，即将若干MapReduce程序以一定语义串接起来形成DAG任务。通过FlumeJava可以高效地完成复杂的任务。
○ Tez是Apache孵化项目，其本身也是一个相对通用的DAG计算系统
第12章 流式计算
• 流式计算（Stream Processing）：在很多应用场所，对大数据处理的计算时效性要求很高，要求计算能够在非常短的时延（Low Latency）内完成，这样能够更好地发挥流式计算系统的威力
○ 优秀的流式计算系统应该具备以下特点
§ 记录处理低延迟
§ 极佳的系统容错性
§ 极强的系统扩展能力
§ 灵活强大的应用逻辑表达能力
• 流式计算系统架构：
○ 主从架构：Storm架构中存在两类节点：主控节点和工作节点。
§ 主控节点上运行Nimbus，其主要职责是分发计算代码、在机器间分配计算任务以及故障检测等管理功能，类似于Hadoop 1.0中的JobTracker的角色。
§ 集群中的每台工作服务器上运行Supervisor，其监听Nimbus分配给自己的任务，并根据其要求启动或者停止相关的计算任务，一个Supervisor可以负责DAG图中的多个计算任务。
§ ZooKeeper集群用来协调Nimbus和Supervisor之间的工作，Storm将两者的状态信息存储在ZooKeeper集群上，这样Nimbus和Supervisor都成为无状态的服务，从而可以方便地进行故障恢复，任何构建发生故障都可在另外一台机器上快速重新启动而不丢失任何状态信息。
§
○ P2P架构（S4）：没有中心控制节点，集群中的每台机器既负责任务计算，同时也做一部分系统管理工作。好处是系统可扩展性和容错性能好，不会产生单点失效问题，但是缺点是管理功能实现起来较复杂。PE（Processing Element）是S4的基本计算单元，属于DAG任务的计算节点，其接收到数据后触发用户应用逻辑对数据进行处理，并可能产生送向下游计算节点的衍生数据。
§
○ Samza架构：在Kafka和YARN之上封装了流式计算语义API的系统，其中，Kafka负责数据流的存储与管理，YARN负责资源管理、系统执行调度和系统容错等功能，Samza API则提供了描述执行流式计算DAG任务的接口。由此可以看出，其本质上是类似于MR 2.0一样运行在YARN统一框架下的具体计算框架。
§
• DAG拓扑结构：一般的流式计算任务都是由计算节点和流式数据构成的DAG有向无环图。
○ 在流式计算系统的DAG拓扑结构图中，计算节点分为两类，一类是整个计算任务的数据输入节点，负责和外部其他系统进行交互，并将输入数据接入流式计算系统，Storm中将这类计算节点称为Spout。第二类节点是完成计算任务的任务计算节点，在Storm中被称为Bolt，整个计算任务就是由若干此类节点通过流经计算节点的流式数据串接起来完成的。每个此类计算节点往往从上游节点接收数据流，对数据流进行特定的计算处理，然后产生衍生数据流，并分发到其下游的计算节点。
○ DAG拓扑结构中的边是由连续不断进入流式计算系统的数据构成的数据流。Storm则将每条数据用数据元组（Data Tuple）来表示，虽然并未明确指明数据主键，但是实际可以将主键放在元组中特定的位置来对主键和其他内容进行区分，接收到数据的计算节点可以从元组对应的内容中读出所需的数据。
○ DAG结构中最常见的基本拓扑结构包含：流水线、乱序分组、定向分组和广播模式。
§ 流水线（Pipeline）是最常见的基本拓扑结构，其将两个计算任务通过数据流连接起来。
§ 乱序分组（Shuffle Grouping）是描述并发的两个计算任务间某种特殊连接方式的基础拓扑结构。计算任务A有两个并发计算节点，计算任务B有三个并发计算节点，如果计算任务A向下游的计算任务B分发数据流时遵循乱序分组机制，即上游节点将其输出的数据随机分发到下游某个计算节点中，则这个结构可被看作是乱序分组的基本结构。乱序分组往往是大数据情况下对数据进行负载均衡的较好机制。
§ 定向分组（Field Grouping）从体系结构上与乱序分组类似，不同点在于上下游计算节点分发数据的模式，定向分组的上游节点在分发数据时，往往根据数据的某个属性（比如主键）进行哈希计算，保证同一属性内容的数据被固定分发到下游的某个计算节点。这种结构对流式计算中类似于数据累加和Join等类型的操作是必需的，因为如果要累加某个Key的数值，不论上游哪个节点输出，都应该路由到同一个下游节点进行累加才能保证数据的正确性。
§ 广播模式拓扑结构也与乱序分组类似，只是其数据分发模式与其不同。广播模式的上游计算节点在分发数据时，同一个数据要向所有的下游计算节点各自分发一次。
• 送达保证（Delivery Guarantees）
○ Storm的送达保证机制：通过“送达保证机制”和“事务拓扑”（Transaction Topology）联合完成的。“送达保证机制”能够实现“至少送达一次”语义，而“事务拓扑”则保证不会出现多次送达的情形。Storm独创的“送达保证”运行机制如下：
§ 数据源节点（Spout）对于每条送入系统内的数据（假设是数据i）赋予一个64位长的消息ID，作为输入数据i的唯一标识
§ 下游节点在接收到数据i及其消息ID后，对数据i进行变换，可以生成0个或者多个其他数据，对于新产生的数据，也分别赋予一个64位的随机值ID。如上所述，每个新数据也会记住其原始输入数据消息ID，以此表明其是由数据i衍生出的。
§ 如果计算节点N成功地接收到了数据i（或者由其衍生的数据），并完成了相应的应用逻辑操作，则通过ACK()函数用异或（XOR）操作来更新表T中数据i对应的签名，即将N的输入数据的随机ID和由这个输入数据产生的所有新数据的随机ID一起与消息i的签名进行XOR操作，用XOR之后的值替换原先的签名数值。
§ 一个新数据可以由多个不同的输入数据共同生成，此时，虽然Storm为这个新数据只生成一个随机ID，但多个标识输入数据来源的数据源ID会绑定到这个新数据上，用来标明其是由这些原始输入数据衍生的
§ 当在某个计算节点更新数据i对应的签名后，如果其签名变为数值0，则说明Storm已经成功地处理掉了原始输入数据i，不会再向下游节点传播数据i产生的衍生数据。此时，Storm向最初产生这条数据的数据源Spout节点发送commit消息告知已成功处理此条数据。
§ Storm会定期扫描系统表T，对那些一定时间内没有被正确处理的消息（即ID对应的签名不为0），则认为在处理这条消息的某个环节产生问题，于是通知数据源Spout节点重新发送该消息。
§ 上述根据数据i的衍生数据被Storm赋予的随机ID不断更新签名的过程中，并不能保证完全的可靠性，因为有可能在数据并未正确处理完之前，碰巧通过XOR得出一个数值0，即数值0并不一定代表数据被正确处理，此可能性太低，几乎不存在。
• 状态持久化
○ 容错的三种模式
§ 备用服务（Standby Service）计算任务的某个计算节点N在另外一台物理机上设置其对应的备份服务S，计算框架定时通过心跳或者ZooKeeper来及时捕获服务状态，当节点N发生故障时，启动备份服务S来接替计算节点N的功能。这是非常常见的一种模式，但是这只适合计算节点属于无状态（Stateless）类型的服务，因为一旦计算节点N死掉，如果存有状态信息，则状态信息全部丢失，无法在计算节点S进行状态恢复。
§ 热备（Hot Standby）热备机制的计算节点N和其备用节点S同时运行相同的功能，上游节点将数据流同时发往下游的计算节点N及其备用节点S，当计算节点N发生故障时对系统无任何影响，因为备用节点S一直和节点N同时运行，所以即使是有状态的服务，两者也时刻保持着相同的状态信息。其好处是显而易见的，但是也有对应的缺点：一个是备用节点额外耗费各种系统资源。另外，正常运行时，在两个节点的下游需要有“流选择器”来保证只有一个上游数据能够通过，避免数据重复。
§ 检查点（Checkpointing）目前大数据处理系统中使用最多的一种类型，为了能够在故障替换时恢复计算节点N的状态信息，计算节点N周期性地将其状态信息通过检查点的方式在其他地方进行备份，当计算框架侦测到计算节点N发生故障时，则启动备用节点S，并从Log中将对应的状态信息进行恢复。两个缺点。首先，如果状态信息较多，为了恢复状态信息，备用节点切换过程可能较长。其次，检查点备份的时间周期也需要仔细斟酌。目前主流的流式计算系统都采用检查点的容错机制，比如Storm、MillWheel和Samza都采用了这一机制。为了保证不会丢失任何状态信息，对于流经的每条数据，只要计算节点处理之后状态信息发生变化都需要进行状态信息备份
○ Storm的状态持久化：Storm使用了“事务拓扑”（Transaction Topology）机制来同时实现状态持久化和“恰好送达一次”语义，其具体机制如下。
§ 将多条数据记录封装成一份批数据（Batch），每份批数据由Storm绑定一个事务ID，事务ID是单调增长数值类型，也即先进入系统的批数据，其事务ID小；后进入系统的批数据，其事务ID大。
§ 在发送这份批数据前，Storm首先通知任务的所有计算节点要开始一项事务意图（Transaction Attempt）。然后Storm将数据送入流式系统中，历经各个应达的计算节点，直到计算结束。最后，Storm通知所有的计算节点该事务意图已经结束，各个计算节点此时可以通过Trident提交其状态信息，也即可以通过事务的方式进行状态持久化。
§ Storm保证每个节点的事务提交顺序是全局有序的，即事务ID编号小的一定在编号大的事务之前提交，此时，计算节点可以执行下列持久化逻辑。
□ 最新的事务ID和此时对应的节点状态信息一起存入Trident，在真正存入之前做下面两个步骤的检查
□ 对该节点来说，如果在Trident中未发现有目前要提交的事务ID，此时可以将事务ID和状态更新到数据库中
□ 如果在Trident中已经发现存在待提交的事务ID，那么Storm会放弃这次提交，因为这说明这次接收到的批数据是系统重发后到达该节点的，而这个节点之前已经成功处理过这份数据，并成功将状态信息在数据库中持久化了，之所以系统会重发，应该是Storm的其他计算节点而非本节点的故障导致的
第13章 交互式数据分析
• Hive系数据仓库：Hive是Facebook设计并开源出的构建在Hadoop基础之上的数据仓库解决方案，Hive能够处理超大规模的数据且有更好的容错性。
○ Hive的本质思想可以看作是：为Hadoop里存储的数据增加模式（Schema），并为用户提供类SQL语言，Hive将类SQL语言转换为一系列MR任务来实现数据的处理，以此手段来达到便利操作数据仓库的目的。
○ 目前对Hive的诟病有很多，主要是其处理效率不够高，这主要是因为Hive和Hadoop的绑定关系太紧密导致的。之所以说Hadoop制约Hive，是因为Hive效率低的主要原因是MR固有的一些特性导致的。
○ 数据组织形式：
§ Hive将存储在HDFS中的文件组织成类似于传统数据库的方式，并为无模式（Schema Less）的数据增加模式信息。除了支持常见的基本数据类型如int、float、double和string外，Hive还支持List、Map和Struct等复杂的嵌套数据类型。
§ Hive的数据组织形式采取分级结构，Table是最基本的数据单元，Table由若干行记录构成，每条记录由若干列组成。其层级结构如下（其中的Partition和Bucket层级是可选的）。
□ Table：每个数据表存储在HDFS中的一个目录下。
□ Partition：一个数据表可以切割成若干数据分片，每个数据分片的数据存储在对应数据表在HDFS相应目录下建立的子目录中。
□ Bucket：数据桶可以理解为将数据表或者某个数据分片根据某列的值通过哈希函数散列成的若干文件，一个文件对应一个数据桶。
□
○ Hive架构
§ Hive提供了类SQL的HiveQL语言来供用户对数据进行相关操作。Hive本质上就是通过数据组织形式为无模式的Hadoop数据增加模式信息，并通过将用户提交的HiveQL语言编译成由MR任务构成的DAG任务图，利用Hadoop的MR计算机制来完成各种数据操作请求。
§
§ 各构件说明:
□ 元数据管理（Metastore）：存储和管理Hive中数据表的相关元数据，比如各个表的模式信息、数据表及其对应的数据分片信息、数据表和数据分片存储在HDFS中的位置信息等。为了加快执行速度，Hive内部使用关系数据库来保存元数据。
□ 驱动器（Driver）：驱动器负责HiveQL语句在整个Hive内流动时的生命周期管理。
□ 查询编译器（Query Compiler）：其负责将HiveQL语句编译转换为内部表示的由MR任务构成的DAG任务图。
□ 执行引擎（Execution Engine）：以查询编译器的输出作为输入，根据DAG任务图中各个MR任务之间的依赖关系，依次调度执行MR任务来完成HiveQL的最终执行。
□ Hive服务器（Hive Server）：提供了Thrift服务接口及JDBC/ODBC服务接口，通过这个部件将应用和内部服务集成起来。
□ 客户端（Client）：提供了CLI、JDBC、ODBC、Web UI等各种方式的客户端。
□ 扩展接口（Extensibility Interface）：提供了SerDe和ObjectInspector接口，通过这两类接口可以支持用户自定义函数（UDF）和用户自定义聚合函数（UDAF），也能支持用户自定义数据格式解析。
§ HiveQL语句可以通过以下方式提交：命令行接口（CLI）、Web UI、满足Thrift接口定义的外部调用或者JDBC/ODBC接口。驱动器在接收到HiveSQL语句后，将其交给查询编译器，查询编译器首先利用元数据信息对语句进行类型检查和语义解析等工作，之后生成逻辑计划（Logical Plan），然后使用一个简单的基于规则的优化器（Optimizer）对逻辑计划进行优化，其后生成由若干MR任务构成的物理规划（Physical Plan）。执行引擎根据这些MR任务之间的依赖关系来调度执行对应的任务最终完成查询，并将查询结果返回给用户。
○ StingerInitiative：Hortonworks公司专门针对Hive性能不足提出的改进。除了增加更丰富的SQL语言支持、自动进行Join操作的优化选择等基础改进外，Stinger还提出了向量查询引擎（Vector Query Engine，充分利用现代CPU架构的L1缓存和流水线，减少程序判断分支及减少函数调用次数，以此来加快查询处理速度，同时，将原先的一次处理一条记录的模式改造为一次处理一批记录的并行处理方式，增加并发性）和基于成本的优化器（Cost-based Optimizer，利用数据本身的特点进行动态优化）
• Shark系数据仓库：Berkeley大学在Spark大数据处理协议栈上建立的支持交互式分析的数据仓库。与底层Spark平台的紧密耦合可能会造成后续更多优化措施的引入困难。
○ Spark是比较适合解决迭代式机器学习问题的，采用RDD的处理模型来对数据进行高效处理。得益于此，Shark很自然地获得了两个优势：
§ 方便地将待处理数据放在内存，所以效率较高
§ 可以通过用户自定义函数（UDF）的方式便捷地加入复杂的机器学习算法
○ Shark架构
§ Shark是能够兼容Hive系统的，其整体复用了Hive的架构和代码，两者整体架构的相似性很高。Shark在以下模块对Hive进行了改写：查询优化器（Query Optimizer）、物理计划（Physical Plan）和执行引擎（Execution）。其大幅度提升性能主要靠以下三个因素
□ 采用了基于内存的列簇式存储方案
□ 采用了“部分DAG执行引擎（Partial DAG Execution，简称PDE）”，本质上是对SQL查询的动态优化，与很多其他SQL-On-Hadoop系统的基于成本的查询优化（Cost-based Optimizer）功能类似
□ 数据共同分片（Data Co-Partition）。在语言层级支持数据共同分片，这有利于提升Join的操作效率。
§
○ 部分DAG执行引擎（PDE）
§ 本质上是一种对查询的动态优化，可以优化Join操作效率，并动态调整操作符运算的并发数。
§ PDE动态收集数据表或者中间数据的统计信息，包括数据表及数据分片的大小、热点数据以及数据分片的统计直方图等信息。
§ 对将要进行的Join操作，可以根据数据表的大小动态选择是以Shuffle Join还是Map Join的方式来实际完成，这样能够提升Join操作效率。
§ PDE可以根据每个数据分片的大小，将很多较小的数据分片合并成较大的数据分片，以此减少Reducer的数目，增加系统的执行效率。
○ 数据共同分片：在数据加载的过程中，两个表在进行数据分片时，根据要进行Join操作的列通过哈希等方法把相同Key的不同表记录内容放到同一台机器中存储，这样后续进行Join操作时可以避免Shuffle等网络传输开销。
• Dremel系数据仓库：作为BigQuery的后台服务，由Google设计开发的超大规模数据交互分析系统，PB级（10亿记录级别）的数据存储在几千台普通的商用服务器上，数据分析人员可以采用类SQL语言对海量数据进行分析和处理，对于大多数查询，Dremel可以在若干秒内返回查询结果。
○ Dremel在数据上快速响应用户查询，主要依赖三点：
§ 架构上借鉴了Google搜索引擎响应用户查询时采用的多级服务树（Serving Tree）结构。
§ 为终端用户提供了类SQL查询语言，不将用户查询转换为MR任务执行，而是通过自身机制（类MPP并行数据库方式）对存储在磁盘中的数据直接进行扫描等数据处理操作
§ 在数据组织形式上采用了针对嵌套式复杂数据（Nested Data）的行列式混合存储结构
○ 服务树结构
§ 最上层一般由一台服务器充当根服务器（Root Server），其负责接收用户查询，并根据SQL命令找到命令中涉及的数据表，读出相关数据表的元数据，改写原始查询后推入下一层级的服务器（即中间服务器）
§ 中间服务器（Intermediate Server）改写由上层服务器传递过来的查询语句并依次下推，直到最底层的叶节点服务器（Leaf Server）。
§ 叶节点服务器可以访问数据存储层或者直接访问本地磁盘，通过扫描本地数据的方式执行分配给自己的SQL语句，在获得本地查询结果后仍然按照服务树层级由低到高逐层将结果返回，在返回过程中，中间服务器可以对部分查询结果进行局部聚集等操作，当结果返回到根服务器后，其执行全局聚集等操作后将结果返给用户。
§
○ PowerDrill：是Google开发的针对大规模数据采用类SQL语句提供查询接口的交互数据分析系统，PowerDrill最大的不同是将待分析的大部分数据加载到内存中进行查询，这决定了PowerDrill的特点：分析速度快，但是处理的数据规模相对有限。
§ PowerDrill有如下三个特点：
□ 采用列式存储
□ 将待查询数据大部分都加载到内存中，这样会明显加快SQL语句的执行速度，通过设计一些精巧的数据结构和更好的数据压缩算法来增加内存利用率
□ 通过将数据记录进行分片并设计一些精巧的数据结构，在SQL执行时，只扫描部分数据（92.4%的记录可被跳过），极大地提升SQL语句的执行速度
○ Impala是Cloudera推出的开源的大数据实时交互式查询系统
§ Impala整体架构
□ 客户端CLI：客户端交互接口，同时也支持Hue、JDBC、ODBC查询接口
□ Impalad：部署在每个数据节点上，接收用户的SQL语句，查询计划器（Query Planner）根据Hive的Metastore中存储的元数据将SQL语句转换为分布式的查询计划，调度器（Query Coordinator）将查询计划分发给存储SQL语句中涉及的数据表数据的其他相关Impalad进程，每个Impalad进程的查询执行器（Query Executor）读写数据来处理查询，并把处理结果通过网络流方式传送回负责该SQL语句的调度器，调度器做全局统计操作后将结果返回给客户端，完成SQL语句的执行。
□ Statestore：通过周期性心跳检测的方式跟踪集群中的所有Impalad进程的健康状况，并将信息动态通知所有的Impalad进程
§ Impala的查询计划：将SQL语句转换为若干可并行执行的计划片段（Plan Fragment）
□ 两个基本目标
® 最大程度地进行并行化
® 最大化数据局部性，即计算离数据越近越好，尽可能减少网络数据传输
• Presto是Facebook开源出的Hive替代产品，主要用于实时场景的交互式数据分析
○ 整体架构：
§ 客户端将SQL查询提交到协调器（Coordinator），协调器根据元数据对SQL语句进行语法检查、语义分析以及并行的查询计划，调度器（Scheduler）将查询计划分配到保存数据表数据的各个工作进程，并监督SQL语句的执行过程，执行结束后将结果返回给客户端。Presto将数据加载到内存进行处理，而且采用MPP并行数据库类似的进程间直接通信的方式来完成查询计划。
§
○ 目前Presto可以支持HDFS、HBase、Scribe等多种数据源
• 混合系数据仓库：出发点是希望能够通过有机集成Hadoop和DBMS，使得整个系统既有Hadoop的高可扩展性和强容错性，又有关系数据库的高效率
○ HadoopDB是建立在Hadoop和Hive基础上的，本质上最终执行的还是一系列的MR任务，所以Hive面临的问题如MR任务启动开销大以及中间结果的磁盘读写仍然存在。但是由于其对于部分聚合类操作和全部Join类操作都下推到数据库节点执行，这样能够充分利用数据库对此类操作的高效率，所以性能也有一定提升。
第14章 图数据库
• 图数据内部存储结构往往采用邻接矩阵或邻接表的方式。图数据与大数据处理中常见的KV数据相比，数据局部性很差，相互之间有很密切的关联，很多自然图的结构遵循Power Law规则（长尾理论），数据分布极度不均匀，极少的节点通过大量的边和其他众多的节点发生关联。数据局部性差意味着数据分布到集群中的机器时存在潜在的数据分布不均匀或者计算中需要极高的网络通信量等问题。
○
• 图数据库分为两类：
○ 在线查询类图数据库：更关注用户查询低延时响应和系统高可用性
○ 离线挖掘类图数据库：更强调数据挖掘等后台处理任务的数据吞吐量及任务完成效率
• 在线查询类图数据库
○ 三层结构
§ 分布式存储引擎层：采用分布式架构，具体使用何种存储引擎没有限制，采用MySQL数据库居多。
§ 图数据管理层：起到以下三个作用
□ 对底层分布式存储引擎的管理功能，包括数据的分片与分发、对查询的路由、系统容错等
□ 图操作逻辑到底层物理存储层读写操作的逻辑转换：将图语义转换为对应的若干SQL语句
□ 优化读操作效率
§ 图操作API层：封装符合图操作逻辑的对外调用接口函数，以方便应用系统在线查询
○ TAO：Facebook的图数据库，保存了所有的实体及其属性、实体关系数据，网站页面的数据读写请求都由TAO来提供服务。
§ 采用数据“最终一致性”的跨数据中心分布式图数据库，为了能够实时响应应用请求，TAO以牺牲强一致性作为代价，系统架构更重视高可用性和低延时，对读操作做了很多优化
§ TAO整体架构：
□ 将多个近距离的数据中心组合成一个分区（Region），每个分区内的缓存负责存储所有的实体和关系数据。
□ 其中，在一个主分区的数据库和缓存中集中存储原始数据，其他多个从分区存储数据副本。缓存对于快速响应用户读请求有巨大的帮助作用，但缓存需要放在内存中，成本高、存储量小，在地域上比较接近的多个数据中心作为一个整体来完整地存储所有的备份数据，在成本和效率上做了权衡和折中。
□ TAO在分区内的存储架构可划分为三层
® 底层是MySQL数据库层，因为数据量太多，将数据分表后形成若干数据切片（Shard），一个数据切片由一个逻辑关系数据库存储，一台服务器可存储多份数据切片
® 第二层是与底层数据切片一一对应的缓存层，称之为主Cache层（Leader Cache），主Cache负责缓存对应的逻辑数据库内容，并和数据库进行读写通信
® 最上层是从Cache层（Follower Cache），多个从Cache对应一个主Cache，负责缓存主Cache中的内容。
□ 缓存设计成二级结构降低了缓存之间的耦合程度，提升整个系统的可扩展性
□
§ TAO的读写操作：
□ 读操作
® 客户端有数据请求时，和最近的从Cache建立联系（无法访问主Cache），如果是读取操作且从Cache中缓存了该数据，则直接返回即可
® 如果从Cache没有命中用户请求（Cache Miss），则将其转发给对应的主Cache，如果主Cache也没有命中，则由主Cache从数据库中读取，并更新主Cache，然后发消息给对应的从Cache要求其从主Cache加载新数据。
□
□ 对于写操作，不论是主分区还是从分区，一定会交由主分区的主Cache来更新主数据库。主数据库更新成功后，主数据库会通过消息将这一变化通知从分区的从数据库以保持数据一致性，也会通知从分区的主Cache这一变化，并触发主Cache通知从分区的从Cache更新缓存内容
§ TAO数据“读你所写”一致性：发出写操作的客户端一定能够读到更新后的新数值而非过期数据
□ 如果数据更新操作发生在主分区，可以保证“读你所写”一致性
□ 从分区的客户端发出写请求
® 从Cache将请求转发给主Cache，主Cache将写请求再次转发给主分区的主Cache，由其写入主数据库
® 写入成功后，从分区的主Cache通知本分区的从Cache更新缓存值，以上操作是同步完成的，尽管此时从分区的数据库可能还未接收到主数据库的更新消息，但是从分区的各级Cache已经同步更新了
• 常见图挖掘问题
○ PageRank计算：得出结果是根据网络拓扑结构分析出的网页重要性评价指标。
§ 该网页PageRank的计算基于以下两个基本假设。
□ 数量假设：在Web图模型中，如果一个页面节点接收到其他网页指向的入链数量越多，那么这个页面越重要。
□ 质量假设：指向页面A的入链质量不同，质量高的页面会通过链接向其他页面传递更多的权重。所以质量越高的页面指向A，则A越重要
§ PageRank算法刚开始赋予每个网页相同的重要性得分，通过迭代递归计算来更新每个页面节点的PageRank得分，直到得分稳定为止
○ 单源最短路径（Single Source Shortest Path）：指定源节点StartV后，求StartV到图中其他任意节点的最短路径
§ 在迭代计算过程中，每个图节点V保存当前自己能看到的到StartV的最短距离，图中的有向边e:(i→j)上带有权值，权值代表从i节点到j节点之间的距离。对于任意一个图节点p来说，首先从上一轮迭代中有入边指向节点p的其他邻接节点中传来的消息中进行查找（每个消息包含指向节点p的邻接节点q自身目前到StartV节点的最短距离和e:(q→p)权值之和），找出外部节点传入的所有距离信息中最短的距离值，如果这个最短距离比节点p自身保存的当前最短距离小，则节点p更新当前最短的距离，并将这一变化通过出边传播出去，也就是将最新的最短距离加上出边的权值传播给节点p有出边指向的节点，代表了从p到这个节点最新的最短距离。通过若干次迭代，当节点的数值趋于稳定时，则每个节点最终保存的当前最短距离就是计算结果，即这个节点到StartV的最短距离。
○ 二部图最大匹配：
§ 二部图是这样一个图G(V,E)：其图节点V可以划分为两个集合Vx和Vy，图中任意边e∈E连接的两个顶点恰好一个在Vx，一个在Vy中。
§ 二部图的匹配，指的是二部图G的一个子图M，M中的任意两条边都不依附于同一个节点。而最大匹配即是指边数最多的那个二部图匹配。
§ 资源的最优分配与任务的优化安排等问题都可以归结为二部图最大匹配问题。典型的解决二部图最大匹配问题的方法包括网络最大流和匈牙利算法等
• 离线挖掘数据分片
○ 数据分片考虑因素：机器负载均衡以及网络通信总量
○ 切边法（Edge-Cut）：切割线只能穿过连接图节点的边，通过对边的切割将完整的图划分为p个子图。
○ 切点法：切割后的图中，每条边只会被分发到一台机器上，不会重复存储，但是被切割的节点会被重复存储在多台机器中，
○
• 离线挖掘计算模型
○ 分为两类：
§ 图编程模型：更多地面向图计算系统的应用开发者
§ 图计算范型：图计算系统开发者需要关心的问题
○ 以节点为中心的编程模型（Vertex-Centered Programming Model）：绝大多数离线挖掘类大规模图计算系统都采用这个模型作为编程模型。
○ GAS模型可以看作是对以节点为中心的图计算编程模型的一种细粒度改造，通过将计算过程进一步细分来增加计算并发性。GAS：信息收集阶段（Gather）、应用阶段（Apply）和分发阶段（Scatter）
○ 同步执行模型是相对于异步执行模型而言的，所有的状态变化只有等到下一轮迭代才可见并允许使用
○ 异步执行模型相对于同步执行模型而言，因为不需要进行数据同步，而且更新的数据能够在本轮迭代即可被使用，所以算法收敛速度快，系统吞吐量和执行效率都要明显高于同步模型。
• 离线挖掘图数据库
○ Pregel：Google提出的大规模分布式图计算平台
○ Giraph：用于大规模图计算的Hadoop开源框架
○ PowerGraph：目前主流图计算系统里效率最高的
第15章 机器学习：范型与架构
• 将迭代式机器学习程序改造为并行架构下运行面临的挑战：
○ 单机版通过共享内存获取的全局参数此时需要并发程序通过网络来存取，而网络的通信效率会低很多。增加通信效率或者减少通信量是关键
○ 分布式环境下，运行在不同机器上的并发程序可能执行速度不统一，使最慢的部分能够逐渐加快速度是关键
○ 较强的容错性，当集群中的机器发生故障时如何保障程序运行的正确。
• 机器学习简介
○ 目的：从数据中自动习得模型，并使用习得的模型来对未知数据进行预测。从数据中学习决策函数f:x→y，这个决策函数将输入变量x映射到输出空间的输出变量y中，即根据输入产生预测。
○ 机器学习包括监督学习、非监督学习、半监督学习以及强化学习等
§ 监督学习：学习系统利用事先标注好的特定训练数据集，得到模型（条件概率分布P(y|x)或者决策函数y=f(x)，条件概率分布或者决策函数描述输入和输出随机变量之间的映射关系）。将之应用在测试数据上进行推理和预测。典型：分类（Classification）与回归（Regression）
§ 非监督学习最典型的问题就是聚类问题，目标是构建划分函数f，通过f将无类标号数据集合划分为k个类别，被划分到同一个类别的数据实例之间有很高的相似性，而不同类别之间的实例具有较低的相似性
○ 损失函数（Loss Function）来度量预测错误程度，损失函数是预测值f(x)与真实值y的非负实值函数。
§ 0-1损失函数
□
§ 平方损失函数
□
§ 绝对损失函数
□
§ 对数损失函数
□
○ 经验风险：模型f(x)关于训练数据的平均损失。经验风险最小化策略认为：经验风险最小的模型就是最优模型
○ 数据并行VS.模型并行
§ 数据并行：将训练数据划分成若干子集合，每个子集合都运行相同的学习算法来进行并行训练过程分别得到局部训练模型，再将局部训练模型融合为全局训练模型。
§ 模型并行：在模型参数巨大、单机无法单独完成整个机器学习算法的建模时，将整个机器学习模型分布到多台机器联合完成训练过程。
• 分布式机器学习范型
○ 三种范型
§ 同步范型：并发程序每一轮迭代都需要在相互之间进行数据同步。但并发程序在各个迭代阶段执行进度不统一，每轮同步必然出现快等慢，造成计算资源的浪费，而且网络通信量较多，也会整体拖慢任务的执行进度。包括MapReduce迭代式计算和BSP模型
§ 异步范型：并发程序之间不需要任何数据同步，任意时刻每个并发程序都可以对全局参数进行读取和更新。计算资源利用率高且整体任务的执行速度快；但某些程序可能在迭代轮数上严重落后，可能会造成最终的计算结果不正确
§ 部分同步范型：介于严格同步范型和异步范型之间，并发程序在满足一定条件时进行同步操作。既能在很大程度上保证程序的正确性，也可以以较快的速度完成任务。包括SSP模型
○ MapReduce迭代计算模型
§ 一轮迭代过程：Map阶段将所有的训练数据划分成子集合，每个Mapper程序结合子集合内的训练数据和全局参数计算局部模型，Reduce阶段收集各个局部模型综合计算出全局模型（常用的综合方法是采取求均值），更新全局参数供下一轮迭代的Mapper使用。在一轮迭代后，终结条件判断模块裁决是否达到迭代终止条件。
§ 效率低主要是由于在多轮迭代之间，以及Map和Reduce两阶段之间存在原始数据和中间数据的大量磁盘读写以及网络传输
○ BSP模型（Bulk Synchronous Parallel Computing Model），具有优秀的健壮性、性能可预测性以及可扩展性
§ 体系结构：
□ <处理器-存储器>资源对集合（简称处理器）
□ 点对点（End-to-End）消息传递通信方式的通信网络，处理器由通信网络连接
□ 处理器之间高效的“路障同步”机制（Barrier Synchronization）
□
§ BSP模型垂直结构由沿着时间轴顺序执行的若干超级步（Super Step）计算过程构成，每个超级步由三个依次执行的不同阶段构成
□ 分布计算阶段：多个处理器并发执行分布计算子任务，在计算过程中仅使用本地可得的局部数据
□ 全局通信阶段：所有的处理器在本阶段进行全局性点对点通信，以便相互间交换所需的数据
□ 路障同步阶段：当某个处理器执行到本阶段时，会一直等待，直到整个系统所有的通信操作结束，为下一个超级步的执行做相应的准备工作。
§ BSP模型优势
□ 方便地利用相邻的超级步之间作为容错设置检查点的时机。
□ 路障机制可以避免分布计算中较常出现的死锁问题
□ 程序的正确性和时间复杂度可以事先进行估计与预测
○ SSP模型（Stale Synchronous Parallel）：典型的部分同步模型。
§ 在SSP模型中，假设存在P个并行程序（Worker），这些并行程序都可以独立地对参数θ产生增量更新μ，这些参数更新满足θ=θ+μ的可累加性，即所有的并行程序各自的更新μ累加后，即可得到完全的更新后参数θ。
§ SSP模型满足“过期有界”（Bounded Staleness）特性，即允许每个并行程序读到过期的更新数据，但是将这种参数的过期性限定在有界范围内。
□ 最快的并行程序和最慢的并行程序最多特定时钟周期，如果超过这个阈值，最快的并行程序需要强制等待最慢的并行程序追赶上来。
□ 当一个运行在c时钟周期的并行程序提交一个参数更新μ时，μ对应的时间戳为c
□ 当一个运行在c时钟周期的并行程序读变量或参数θ的值时，它可以看到针对θ的所有时间戳小于或等于c-s-1的参数更新μ，这点由SSP模型来保证。
□ 读你所写一致性（Read-Your-Writes）：一个并行程序p总能看到它自己产生的参数更新μp。
§ SSP模型可以被认为是一种处于同步模型和异步模型之间的折中模型，这样会很好地平衡算法的执行速度与其可收敛性或正确性的关系，于是它既有异步模型的高速执行优势，也有同步模型的可被证明的正确性保证。
• 分布式机器学习架构
○ MapReduce系列：
§ 直接构建在Hadoop平台上的机器学习框架
□ 架构：
® HDFS作为训练数据和机器学习模型的存储场所
® 底层是Hadoop提供的MapReduce计算机制
® 中间层的常用机器学习算法库
® 最上层往往分为模型训练和在线服务两类功能模块
® 代表性的包括Cloudera Oryx系统和Apache Mahout系统
□ Oryx运行流程为：首先将训练数据存入HDFS指定的目录下，计算层根据配置文件内容读取训练数据，然后使用特定参数配置的机器学习算法学习模型，并将习得的计算模型存入HDFS指定的目录下，这样即可完成训练过程。服务层加载习得的模型即可对外提供在线预测功能。
§ Hadoop平台改造的计算框架：包括Twister和Haloop
□ 三点改造内容：
® 消除MapReduce在各个阶段的中间结果磁盘输入／输出以及Shuffle过程的密集网络传输过程
® 将运算中间结果在内存中缓存起来供后续迭代在此基础上持续进行。
® 将数据分布到多机环境时，尽可能将数据和机器之间的分配关系固定化，避免多轮迭代中反复、频繁的网络传输操作。
○ Spark：AMPLab实验室推出，其最核心的部分是适合解决迭代式机器学习类问题的DAG批处理系统Spark，在此基础上，逐渐在其上层开发出流式计算系统D-Stream、图计算系统GraphX、机器学习库MLlib以及MLBase等适用于不同场景的子系统，遂形成了一整套大数据处理技术方案。
§ Spark针对工作集数据（迭代计算过程中会反复重用的很多中间数据）提出了基于内存的分布式存储抽象模型：可恢复分布式数据集（Resilient Distributed Datasets，RDD），这样工作集数据可以有选择性地被加载并常驻在内存中，有利于后续迭代计算过程中大大提升此类问题的处理效率。
§ Spark是集成了RDD模型的DAG批处理系统，在RDD增加数据复用与系统处理速度的优势基础上，同时还具备传统DAG系统很强的容错性、数据局部性感知的调度策略以及高可扩展性，其处理迭代式机器学习任务的效率比MapReduce原始方式快20倍左右。
§ RDD由若干只读的分片数据记录组成，其特点如下
□ 由只读的数据记录组成的数据集合，即RDD一旦生成，其内容是不可更改的，如果数据转换后会形成新的RDD，并记载两个RDD的生成关系——血统记录（Lineage）。
□ RDD是一种粗粒度的数据处理模型，将RDD作为一个整体，对RDD的同一个数据转换操作要应用到RDD内包含的所有记录内容，即每个记录都执行相同的数据转换操作。
□ 对RDD的数据处理可以分为两类：
® 数据转换类（Transformation）：操作对RDD的记录内容发生了更改并形成新的RDD
® 行为类（Action）：RDD需要返回给应用对应的处理值或者是记录写入磁盘等非更改性操作
□ 用户可指定哪些RDD被缓存到内存中
□ RDD的容错采取根据血统记录恢复的方式。
○ MLBase：集成了很多机器学习算法库的分布式机器学习运行框架，并为普通应用者提供了尽可能简单的使用接口
§ 整体架构：采取Master-Slave结构
□ 用户使用MLBase提供的任务声明语言来描述机器学习任务，并提交给MLBase主控服务器（Master）。
□ MLBase将用户请求解析为逻辑学习计划（Logical Learning Plan，LLP），其描述了机器学习任务的一般工作流
□ LLP的搜索组合空间包括各种机器学习算法、算法参数组合空间、数据特征组合空间等，其形成的搜索空间异常巨大。优化器（Optimizer）可以通过一定的策略使搜索过程在一定时间内完成，并找到问题的较优解，形成优化的逻辑计划。
□ MLBase将优化的逻辑计划进一步转化为物理学习计划（Physical Learning Plan，PLP）以供实际执行，PLP由若干机器学习操作符构成。
□ 将物理学习计划分配给各个工作服务器（Slaves）来并行执行，并把执行结果返回给用户
□
• 参数服务器（Parameter Server）：可看作是传统的共享内存方式在网络环境下的并行扩展版本。
○ Petuum：CMU提出的通用参数服务器架构
§ 架构
□ 一台参数服务器充当主控服务器（Name-Node）的作用，并负责数据路由以及数据分片在不同的服务器间分配等工作
□ 参数服务器集群是一个类似于分布式共享内存的分布式KV存储池，用于存储机器学习任务中各个并行客户端共享的全局参数，不同应用的全局参数可以放置在参数服务器集群不同的表格中。
□ Petuum采取部分同步范型，不同的表格可以绑定不同的部分同步参数设置。客户端可以在合适的时机更新参数服务器中对应表格的全局参数，当其更新参数服务器对应的参数后，这个更新后的参数即对其他客户端可见。
○ 一致性模型
§ 分布式机器学习一致性解决方案
□ 序列一致性（Sequential Consistency）：强一致性模型，GraphLab，可以保证算法的正确性，但是严重限制了整个系统的并发程度
□ 完全异步、无一致性的保证：YaooLDA，并发程度高，但是对于算法的正确性没有理论保证
§ 参数服务器架构下的一致性模型
□ 时钟界异步并行（Clock-bounded Asynchronous Parallel，CAP）：当行进快的并行程序比慢的并行程序快太多时（超出时钟范围），则行进快的并行程序需要被阻塞来等待行进慢的并行程序追上来
□ 值界异步并行（Value-bounded Asynchronous Parallel，VAP）：某个并发程序试图更新一个“非同步局部更新”累加值超过Vthr的参数时，系统会阻塞该并发程序，并将这个参数足够多的更新对所有其他的并发程序可见，只有这样才允许该并发程序继续前行
○ SSPTable：SSP模型的一个具体实现架构
§ 架构：典型的Client-Server结构
□ Server是由多台服务器构成的分布式参数服务器集群，用来存储客户端共享的全局参数
□ Client分布在集群的其他机器上：分为进程和线程两级结构。一个客户端代表一个客户进程，内部又包含若干线程。同时客户端维护一个进程缓存，而每个线程各自维护一个自己用的线程缓存，缓存用来暂时存放从参数服务器获取到的数据。
□
§ SSPTable提供了以下三个简洁的API调用接口
□ read_row(table,row,s)：读取某表的某行，指定数据过期阈值为s。
□ inc(table,row,el,val)：将某表某行数据的值增加val，增量可以是负值。
□ clock()：线程通知参数服务器自己的时钟向前迈进一步，并将inc操作的所有更新传给参数服务器。
第16章 机器学习：分布式算法
• 计算广告：逻辑回归
○ eCPM（Effective Cost Per Mille）代表每千次展示可获取的收入。BidPrice为广告商对竞价关键词的出价，出价越高，排名越靠前，CTR（Click Through Rate）为广告的点击率，用广告点击量和展示次数的比率来计算，其表征了广告和用户需求的匹配程度
§
§ BidPrice已事先确定，所以最关键的是要估算对于当前查询而言某个广告创意的CTR。
○ 逻辑回归（Logistic Regression，LR）
§ 二项逻辑回归模型是一种分类模型，其可由条件概率分布P(y|X)来表示，其中，随机变量X的取值范围为实数，随机变y的取值为-1或者1，W是特征权重向量。一般采用如下公式表示：
□
§ 计算广告的应用环境下，y=1代表用户会点击广告，y=-1代表用户不会点击广告，变量X代表<查询，广告>数据对，W是这些特征的权重向量，表明对应特征的重要性，需要训练数据获得。
第17章 增量计算
• 增量计算模式
○ 增量计算流程：
§ 变化传播模式：设计技术方案时，更多地从新数据和受影响的旧数据这个角度来考虑如何设计系统
□ 先计算新数据的结果，然后判断直接受到影响的旧数据有哪些，并重新计算其结果
□ 将这些变化的结果通过数据之间的结构传播出去，再考虑又有哪些其他旧的计算结果会进一步受到影响，如果影响足够大，那么需要重新计算，如此不断循环往复，即可完成整个增量的计算过程。
§ 结果缓存复用模式：更多地从哪些旧数据的计算结果没有发生变化的角度考虑，并在此基础上对数据或者计算流程进行组织，尽可能最大化地复用没有变化的旧的结果，其复用方式往往采用结果缓存，将可复用的旧数据计算结果缓存在内存或者外存文件中。
○ 时效性分类
§ 准实时增量计算：在分钟级别实现增量计算过程，“变化传播模式”
§ 批处理增量计算：没有时效性要求，“结果缓存复用模式”
○ Hadoop平台下增量计算
§ 基础架构
□ 增量计算系统首先要区分哪些数据是新增数据，对于新增数据，需要进行完整的Map和Reduce两阶段的运算，对于旧数据，对Map阶段输出中间结果复用，只进行Reduce阶段的运算
□ 为了最大化复用数据，在Map过程的数据输入阶段，尽可能将新数据和旧数据明确区分开；将上一轮Map运算的中间结果放在一起
• Percolator：Google构建在Bigtable上的与MapReduce互补的增量计算模式，主要用来对搜索引擎的索引系统进行快速增量更新（局部更新）
○ 提供了支持ACID“快照隔离”语义的跨行跨表事务，一个操作涉及更改不同表的不同数据，那么这些更改要么同时生效，要么同时失效
§ 快照隔离维护了数据的不同版本，不同的操作针对不同的数据版本进行，以此来增加并发程度并保证数据的修改一致性