### 1 奇怪的现象 在使用Hazelcast的Eviction时，发现观察到的现象与想象的不同。按照官方文档介绍，Eviction有这样几个配置选项： ~~~ ~~~ 看后三项，按照参数的描述应为：当每个结点的entry数达到5000时，使用LRU策略，剔除25%的entry，即1250个。可是观察到的现象一般是entry数达到4200左右就止步不前，继续大量高并发的insert测试，既不会增长也不会减少。这是怎么回事？而且相比Redis，Hazelcast在达到eviction临界条件后继续并发插入和读写时，性能表现依旧良好，就像没有发生eviction一样。是使用了多线程还是什么神奇的算法？源码之前，了无秘密，还是从代码中寻找答案吧。 ### 2 代码剖析 我们客户端使用的，也是最顶层的API，就是IMap了，这里以比put()更为高效的set()作为分析的起点。后面其实能够看到，因为使用了命令模式（command），两者都是继承一个父类，evict都是在一个地方触发的。 ### 2.1 MapProxyImpl包装器 IMap的直接实现类是MapProxyImpl，但它只是个Wrapper，负责转换key和value。  ### 2.2 MapProxySupport封装命令对象 真正的实现都是在它继承的MapProxySupport类中，例如set()调用的setInternal()就能在Support类中找到。各种internal方法将操作包装成Operation类，这方便了远程调用的实现，例如set()要执行的结点不在本地。  ### 2.3 SetOperation命令模式 invokeOperation()中会确定操作应该在哪个分区执行，这个分区位于哪个结点上。这里Hazelcast维护了一个线程池，每个分区都有对应的线程去执行本分区的操作。因为过程比较复杂，所以这里直接略过，继续关注我们重点想知道的eviction过程的实现。那么直接看一下SetOperation中的逻辑。SetOperation很简单，直接调用RecordStore保存键值对，但afterRun()中有一些隐含的后处理。  ### 2.4 BasePutOperation触发eviction 果然，在afterRun()中除了广播事件、使Near缓存失效外，还有触发eviction过程。Evict()调用的就是RecordStore的evictEntries()方法。  ### 2.5 AbstractEvictableRecordStore控制eviction 真正的evict控制逻辑就在这里。首先，shouldEvict()会判断是否满足了我们之前配置的eviction的触发条件，如PER_NODE=5000。如果满足则调用removeEvictableRecords()开始剔除数据。  ### 2.6 EvictionOperator事有蹊跷 最终removeEvictableRecords调用的是EvictionOperator，具体的实现都在这里。但仔细看这段代码却看不出有什么高明之处，只是简单地迭代RecordStore的记录，将满足条件的entry剔除掉。既没用多线程，也没什么特殊的算法，这到底是怎么回事？  ### 2.7 答案揭晓 谜底其实就在Operation类对应的RecordStore初始化上。我们知道，默认情况下，Hazelcast将map分为271个partition。其实RecordStore也是按这些partition划分的，而不是使用一个大的RecordStore。所以从BasePutOperation的evict()到后续处理的都只是当前key对应分区的RecordStore。也就是说：当key要被处理时，Eviction发生在对应的partition里，而不会evict所有数据的25%（Redis就是处理database中的所有数据，所以延迟会有所增加）。所以，当我们继续压力测试时，不断有key继续插入，这些分区就会不断发生eviction，导致整体的内存使用会保持不变。  ### 3 类图全貌 梳理了上面的执行流程后，我们最后整理一下这些类之间的关系。 

目前Redis Cluster仍处于Beta版本，Redis 3.0将会加入，在此可以先对其主要功能和原理进行一个预览。参考《Redis Cluster - a pragmatic approach to distribution》。 ### 1 没有集群的Redis 没有集群功能的Redis，每个master-slave主从复制都独立于其他结点，sharding需要在客户端如Jedis中控制。可以使用官方提供的Sentinel监控主从的状态，实现自动的Fail-over切换。具体请参见[《Redis主从和HA配置》](http://blog.csdn.net/dc_726/article/details/11694437)。 ### 2 集群拓扑 所有结点直连其他结点，端口为baseport(6379)+4000。为了带宽和性能，通信协议是二进制的。客户端与结点之间的通信还是正常的ascii协议。  虽然结点是互联并且功能等同的，但实际上结点还是分为master和slave两种。例如下图所示，每个master有两个副本，副本不接受写请求。Redis-trib集群管理器会分配master和slave，使其尽量在不同的物理机上。  ### 3 请求处理 Redis集群客户端分为两种：Dummy和Smart： Ø  Dummy模式：单连接，随机连接一个结点，对现有客户端代码结构影响最小。 Ø  Smart模式：长连接到许多结点，在客户端缓存一份hashslot=>node的路由表，当接收到服务器的**-MOVED**响应时更新表项。这种方式具有低延迟，但当集群很大时，客户端会维护许多连接，此时应当共享client对象实例。  当添加新结点，可以使用redis-trib的**MIGRATE**命令进行re-sharding。

TwemProxy是一个Redis的中间件代理，具有很多[有用的功能](http://cloudaice.com/twemproxy-explore/)，可以暂时替代一部分Redis Cluster的功能： ²  支持**失败节点自动删除** ²  可以设置重新连接该节点的时间 ²  可以设置连接多少次之后删除该节点 ²  该方式适合作为cache存储 ²  支持设置HashTag ²  通过HashTag可以自己设定将两个KEY hash到同一个实例上去。 ²  减少与redis的直接连接数 ²  **保持与redis的长连接** ²  可设置代理与后台每个redis连接的数目 ²  **自动分片到后端**多个redis实例上 ²  多种hash算法 ²  可以设置后端实例的权重 ²  避免单点问题 ²  可以平行**部署多个代理**层，client自动选择可用的一个 ²  支持状态监控 ²  可设置状态监控ip和端口，访问ip和端口可以得到一个json格式的状态信息串 ²  可设置监控信息刷新间隔时间 ²  高吞吐量 ²  连接复用，内存复用。 ²  **将多个请求组成redis pipelining**统一向redis请求 ### 安装问题解决 TwemProxy的tarball分发包在Google Code上，无法下载了…只能从GitHub上下载源码包进行手动编译了。安装、升级了autoconf(2.64以上，下载地址：[http://ftp.gnu.org/gnu/autoconf/autoconf-2.69.tar.gz](http://ftp.gnu.org/gnu/autoconf/autoconf-2.69.tar.gz))、automake、libtool后，却一直无法编译安装成功。不是报“Error : possibly undefined macro: AM_INIT_AUTOMAKE”，就是报“Cannot find install-sh, install.sh, or shtool”。最后终于找到[解决办法1](http://toto-share.com/2012/09/error-possibly-undefined-macro-am_init_automake/)和[解决办法2](http://toto-share.com/2012/09/configure-error-cannot-find-install-sh-install-sh/)。 最后，我确定在我环境中能成功编译安装的方法是： >tar -xzvf twemproxy-0.4.0.tar.gz >**aclocal *(****解决错误1)*** >autoconf -f -v –i >**autoreconf -f -i -Wall,no-obsolete *(****解决错误2)*** >./configure --enable-debug=full *(可以开启O3优化：CFLAGS="-O3 -fno-strict-aliasing" ./configure)* >make >src/nutcracker –h *(**查看各种选项)* ### 配置和运行 conf/nutcracker.yml是默认使用的配置文件，打开后能看到配置了alpha, beta, gamma, delta, omega五个连接池作为例子。我们只保留alpha，并配置两个redis服务器，端口为6379和6479。之后相应地，配置好两个Redis实例并启动。现在就可以启动twemproxy了，直接src/nutcracker就可以运行。 测试一下是否连通了。用redis-cli -p 22121连接到twemproxy的监听端口，执行一些set命令，然后连接到两个Redis实例中就能看到有一些key-value保存进去了。

开源IMDG通常都提供了SPI或其他接口，供用户自行扩展。以Hazelcast为例，我们可以用一些好玩的小工具增强其查询、Map和后端持久化的功能。这些小工具虽然看起来很小，但功能也非常强大。  ### SQL查询 JoSQL非常简单易用，只需几步就可以在普通Java对象上实现SQL查询功能，而且对标准SQL支持的还很全面呢。同时也提供了接口，我们可以自定义想要的SQL函数。  它与Hazelcast的集成方法非常简单，就是新建一个Predicate子类。查询时使用我们新建的这个类，Hazelcast执行时会将我们新定义的apply发送到各个结点上执行，将符合条件的数据返回并汇总，最后返回给我们。利用JoSQL对Hazelcast查询的where过滤条件增强比较容易，但是想利用JoSQL的groupby、join和各种汇总函数对Hazelcast进行增强的话改动比较大，因为Hazelcast的设计就是进行简单的过滤查询，返回值都是对象。要扩展的话可能要利用Hazelcast的MapReduce和Aggregator接口。  ### 堆外存储 JDK中提供了直接申请堆外内存的方法，就是ByteBuffer.allocateDirect()，其底层调用的是Sun私有的Unsafe.allocateMemory()。因为像Hazelcast这些开源IMDG都把堆外存储缓存数据作为商业版的主打功能之一，所以我们需要自行扩展。但是自己管理堆外内存很麻烦，涉及到空闲内存分配、回收、内存碎片等等管理问题，所以我们还是希望直接使用第三方的产品。在这一领域，目前已经有一些开源产品了： Ø  MapDB：嵌入式数据库，提供文件和堆外存储，以及ACID支持。 Ø  DirectMemory：Apache旗下的开源项目，目前还是0.3版本。 Ø  其他：HugeCollections、Fast-serialization等。 以MapDB为例，看一下它的使用方法。具体与Hazelcast的集成方法稍稍麻烦一些，需要实现三个Hazelcast的类，可以参考GitHub上的项目：Hugecast和mapdb-hz-offheap。  ### 后端持久化 使用IMDG时，我们通常还要为每个对象的缓存提供数据加载和持久化类，从而实现缓存的初始化加载和read/write-through功能。可以简单地使用JDBC+DozerMap实现数据的加载保存和实体对象装配，或者利用Hibernate简化这个过程。每种IMDG都提供了对应的扩展接口，例如Hazelcast的扩展点是MapLoader或MapStore，具体实现方式很简单就不列举了。

本文对Redis如何保存关系型数据，以及如何对其匹配、范围、模糊查询进行举例讲解，其中模糊查询功能基于最新的2.8.9以后版本。 ### 1 关系型数据的存储 以Staff对象为例，在关系型数据库或类似GridGain的内存网格产品中(底层使用H2数据库的内存模式存储)，我们以表形式保存对象的数据。因为内存网格是基于对象做缓存的，所以还要额外多出一列(Staff列)保存整个对象的编码，例如序列化后的二进制或者JSON格式等，便于直接返回给应用后进行反序列化。而在Redis中，我们可以用id作为唯一标识，使用**key-value、hash、zSet**三种数据结构进行保存。Key-value是为了保存id和整个对象，确定id范围后可根据它将对象返回给客户端，而选择其他两种数据结构的具体原因后面再说。  ### 2 匹配查询 利用hash表的**hget**或**hmget**可以实现dept='IT'或者dept in ('IT', 'QA')这种单值或多值的完全匹配查询。拿到id列表后，再去查询key-value获得到对象。 ### 3 范围查询 因为我们将age保存成zSet的score，value是id，所以可以利用zSet的**zrangeByScore**方法获得score在某一区间范围内的value值。 ### 4 模糊查询 Redis 2.8.9后zSet加入了一个非常有用的方法**zrangeByLex**，我们将score都保存为0，value是姓名:id的格式，利用zrangeByLex可以获得字母在某一区间内的value值。例如，zrangeByLex name [A, (F，可以查询出Allen, Aaron, Carter。 ### 5 分页查询 同时，**zrangeByLex**还支持分页查询，语法类似limit start, offset。 ### 6 局限性 上述举例说明了几种常见查询在Redis的实现方式，但是Redis毕竟只是key-value存储，所以有很多局限性。例如，1）无法实现多条件组合的查询，例如age>25 AND name like 'A%'，硬要实现的话需要多条命令并计算并集或交集。2）模糊查询中文比较费劲： 

作为另一款主流的开源数据网格产品，GridGain是Hazelcast的强有力竞争者。同样提供了社区版和商业版，近日GridGain的开源版本已经进入[Apache孵化器项目Ignite](http://ignite.incubator.apache.org/)(一款开源的内存计算(In-Memory Computing)IMC中间件)，目前Apache正在迁移GridGain开源版本的代码到Ignite项目。鉴于经过之前Hazelcast的介绍已经对数据网格产品有了一定了解，本文着重介绍GridGain与Hazelcast差异化之处。 ### 1 重叠功能列举

比较		Hazelcast	GridGain
使用性	安装	Maven引入Jar包即可，无需安装软件
	客户端	支持各种语言的客户端
	框架集成	集成Hibernate、Web Session、Spring
基本功能	分布式计算工具	分布式的集合、并发包、消息队列、调度器
性能	性能配置	内存索引、Near-Cache、数据亲和性
可靠性	数据备份	分区数据冗余备份
	持久化	read-through，write-through/behind
	事务	保证数据一致性
扩展性	自动分区	支持本地、分区、复制三种方式
	动态拓扑	动态添加删除结点，自动rebalance

上面简单列举了一些Hazelcast和GridGain的重叠功能，而两者的差异之处主要在于以下几个方面： Ø  整体功能的全面性：围绕内存计算提供的**功能**。 Ø  使用性：对**SQL支持**的完整性、对**Continuous Query**的支持、以及与持久化存储的**数据集成**。 Ø  性能：免费的**off-heap存储**实现。 Ø  可靠性方面：**事务**的隔离性、内存**溢出到磁盘**。 Ø  管理：提供强大的**后台管理界面**。 此外，企业版还提供了Portable跨平台对象、安全和审计、数据中心复制、可还原的本地cache以及split-brain网络分段问题解决等功能。本文暂不关注企业版中的附加功能，下面开始着重介绍上面列举的开源社区版与Hazelcast的功能差异。 ### 2 全面的内存计算功能栈 从整体功能上来说，GridGain是个出色的多面手，不仅可以完成本职工作-内存计算/数据网格，还提供了： 1)     GGFS(GridGain In-Memory File System)，类似Spark生态圈中的Tachyon，能够加速MapReduce任务的执行。 2)     完整的ACID和事务支持，可以作为内存数据库。 3)     流式数据/事件处理，可以作为CEP事件处理器。   ### 3 丰富的查询功能 一般开源IMDG产品支持基本的对象过滤查询能力，但GridGain底层借助H2数据库引擎来解析和执行SQL，所以支持复杂的对象联结查询，类似于GemFire中的OQL提供的功能，以及仅在Hazelcast商业版本中才支持的Continuous Query功能。 首先来看一下我们的测试数据和Entity是什么样子，代码忽略了构造函数、getter/setter和toString等方法。Person中包含了id、name、salary三个基本属性，并与Organization是多对一的关系，与Address是一对一的关系。  对于POJO要注意几点：1）查询中涉及的成员变量都要标上**@GridCacheQuerySqlField**注解；2）因为POJO会被哈希到其他结点上的分区，所以要实现**序列化接口**；3）下面例子只测试了**one-to-one**(直接嵌套实体Address)、**many-to-one**(通过orgId关联其他实体)关系的查询，而没有尝试one-to-many和many-to-many(都是在实体中嵌套另一实体id的集合)；4）GridCacheConfiguration要开启**setQueryIndexEnabled(true)**。 #### 3.1 简单的过滤查询 简单的对象过滤查询是最常见的，也是其他网格产品像Hazelcast支持的。  #### 3.2 同cache下的join查询 同cache下可能会关联的数据，可以通过数据亲和性设置使相关数据分配到同一分区中，从而避免网络传输开销。当然，Hazelcast也是支持数据亲和性的，本节关注的重点是join查询。代码类似于3.3中的跨cache查询，差别只不过是：1）Organization对象不是保存到org-cache，而是与Person对象一起保存到person-cache；2）SQL中不需要显式指明缓存名称，因为对象都在一个缓存person-cache中。 #### 3.3 跨cache的join查询 ~~被join的cache(org-cache)必须是REPLICATE的，从而在各个结点上都存在，不会产生交叉join。~~(注：Impala支持这种join，将产生N*M次数据通信)  #### 3.4 字段查询 GridGain不仅支持查询结果为实体，同时也支持各种SQL函数对实体进行各种操作，如聚合、字符串操作等。  #### 3.5 Continuous查询 Continuous查询不支持SQL，只支持Predicate风格组装查询。  执行效果如下，首先初始化到缓存中的Person对象中有一个salary=300，满足条件，所以本地callback收到通知。之后我们试着更新缓存中的一个Person对象的salary=250，于是再次收到通知。最后我们新建一个Person，salary=500，并保存到缓存中，于是再次收到通知。这就是Continuous Query的运行效果。  ### 4 集成持久化存储 类似Hazelcast，GridGain也提供了**read-through**、**write-through**以及异步**write-behind**三种与后端持久化存储通信的方式。此外，GridGain还支持事务提交时批量write，以及缓存entry**即将过期时**自动重新re-cache(**refresh-ahead**)功能。像refresh-ahead功能在GemFire等商业产品中才会实现。  ### 5 堆外内存存储 不像Hazelcast等开源产品只在商业版中提供off-heap存储功能，GridGain在开源版本中就提供了此功能，从而显著地扩充JVM管理的内存容量，并减轻GC压力和停顿时间。GridGain提供了*ONHEAP_TIERED*(**默认堆优先，溢出到非堆内存**)、*OFFHEAP_TIERED*(不使用堆，直接将所有entry放入非堆内存)、*OFFHEAP_VALUES*(将key存储在堆，value存储在非堆内存)三种模式。当堆和非堆内存都不足时，还可以开启SWAP，将数据溢出到磁盘(详见第7部分：数据溢出到磁盘)。下图表示了堆、非堆、磁盘、外围存储的容量和延迟的关系。  ### 6 完整的ACID和事务支持 内存事务与传统数据库事务有一点点不同。因为IMDG产品使用的是易逝内存，所以故障或断电时内存数据会全部丢失。一般IMDG重启时会从备份结点或其他持久化存储中恢复数据。但这不代表内存事务不重要！只要集群是存活的，GridGain就要保证不同结点间的数据一致性。为此，GridGain提供了两种*TRANSACTIONAL*和*ATOMIC*两种配置： Ø  **TRANSACTIONAL**：完整的ACID属性的事务，以及显式的锁。 Ø  **ATOMIC**：没有事务和锁。 GridGain使用**2PC(两阶段提交)协议**实现分布式事务，同时支持**乐观**和**悲观**两种模式。乐观模式下所有key在提交时才会加锁，所以在Prepare阶段，prepare消息发送给各结点获取事务中将要操作的key的锁，各结点通过ACK消息应答。而悲观模式下，所有key在提交前就已加锁，所以Prepare阶段不需要做任何事。在Commit阶段，commit消息发送给各结点提交，若失败则发送回滚消息给各结点。可以设置各个结点之间是同步还是异步提交(注：Hazelcast支持一种**2PC扩展协议**，具体的优势还有待研究)。最后，GridGain支持*READ_COMMITTED*，*REPEATABLE_READ*和*SERIALIZABLE*三种事务隔离级别，默认是REPEATABLE_READ。而Hazelcast只支持REPEATABLE_READ一种。 ### 7 数据溢出到磁盘 在第5部分堆外内存存储中提到过GridGain的层次化存储，以下面的缓存配置为例，它同时开启了off-heap和swap。1）首先，数据优先保存在堆内存中。2）当entry总数超过100个时，会通过LRU淘汰到非堆内存中。3）当超过非堆内存的最大容量5MB时，会将多出的数据保存在磁盘上。当然，磁盘IO操作的代价是很大，我们可以优先考虑使用超大的**非堆内存**，或者使用**SSD闪存**，以及开启操作系统的**磁盘IO缓存**来进行优化。 GridCacheConfiguration cacheCfg = new GridCacheConfiguration(); cacheCfg.setEvictionPolicy(new GridCacheLruEvictionPolicy(**100**)); cacheCfg.setOffHeapMaxMemory(**5 * 1024L * 1024L**); cacheCfg.setSwapEnabled(**true**); ### 8 强大的管理界面 首先，以相同配置启动三个GridGain实例。然后，启动gridgain-fabric-os-6.5.5/bin/ ggvisorui.exe，在Visor GUI中选择File->Connect->External标签页下，直接localhost+默认端口连接即可进入Dashboard。在这能看到我们刚刚启动的三个结点的总体信息。  点击Data Grid标签页，可以查看各个cache的内存使用、读写以及命中情况，例如我们初始化了3个Person和2个Organization对象，并分别保存到了person-cache和org-cache中，于是我们可以在此标签页看到Primary Entry和Write个数都是3和2。 

数据网格产品经常会使用P2P进行通信，借此机会系统地学习一下P2P网络和其资源搜索策略。 ### 1 P2P网络架构 谈到P2P就涉及到一个概念：**Overlay Network(覆盖网络)**。所谓覆盖网络是应用层网络，几乎不考虑网络层和物理层，它具体指的就是建立在另一个网络上的网络。例如P2P网络就是覆盖网络，因为它运行在互联网之前，但允许对未知IP主机的访问。通过DHT等算法，可以在事先不知道IP地址的情况下，访问到存储某个文件的结点。  常见的P2P系统主要有Unstructured Network和Structured Network两种架构。 ### 1.1 Unstructured Network 非结构化的P2P网络并不给覆盖网络强加某种特定架构，而是结点间随机形成链接。最流行的P2P网络，像Bittorrent、eMule、Gnutella、Kazaa等都是非结构化的P2P协议。因为缺少结构，所以网络面对频繁的动态添加和删除结点时，依然能够健壮地运行。但也正因为缺少结构，所以当某个结点想要搜索某些数据或文件时，查询必须flood整个网络(详见1.3搜索策略)。  ### 1.2 Structured Network 结构化P2P网络将覆盖网络组织成某种特定的拓扑结构，并且它的协议能够保证：**任何结点都能高效地搜索网络中的资源**，即使资源是非常冷门的。常见的结构化P2P网络通常实现**一致性哈希**或者其变种**分布式哈希表DHT**分配文件的所有权到特定的结点。  ### 1.3 搜索策略 两种P2P网络可以使用不同的检索策略：中央索引、本地索引和分布式索引。分布式索引策略是对其他两种策略的权衡。  #### 中央索引(中央服务器) 由一个中央服务器统一保存资源与结点的关系。这种方式搜索效率比较高，因为可以通过中央索引直接定位到目标结点，然而这种方式有时并不可行，特别是集群规模特别大时。  #### 本地索引(flooding搜索) 每个结点保存自己的资源信息，当寻找不属于自己的资源时，会flooding整个网络进行寻找。这种flooding的方式会在网络中引起大量的traffic，并使每个结点都要处理查询请求而导致高CPU和内存使用率。并且flooding不保证通信质量，所以flooding也无法保证一定能够找到保存指定数据的那个结点。因为热数据在多个结点上存在，所以比较容易搜索成功。反之，冷数据只在很少的结点上存在，所以搜索很可能会以失败告终。并且搜索效率也很低，也容易导致安全问题。  #### 分布式索引 为了高效地在网络中搜索，结点需要保存满足特定条件的邻居的列表，这使得整个网络在高频率的添加删除结点时，没有非结构化网络那样健壮。使用DHT路由的结构化P2P网络的著名软件有BitTorrent，Kad Network，以及各种研究项目Chord等。**基于DHT的网络在网络计算系统中也有广泛的应用，它帮助实现高效的资源发现和应用程序调度等**。 典型的P2P网络使用的架构和搜索策略如下：  ### 2 基本的分区算法 经典的分区算法有直接取模和哈希取模。假设有K台服务器，我们可以这样确定数据X所在的服务器i： Ø  直接取模**i = X mod k**：问题是数据分布不均匀。 Ø  哈希取模**i = hash(X) mod k**：问题是1）根据集群和数据规模，散列冲突可能会比较严重(只能通过替换更好的哈希算法来平衡)；2）扩容添加结点或故障删除结点时(k->k±1)，所有数据都要重新映射到新的结点上(通过后面介绍的两种分布式哈希可以解决)。  ### 3 一致性哈希 ### 3.1 构造 **将结点和数据映射到同一个线性地址空间**，结点负责保存前一结点到本结点之间的数据。  ### 3.2 Lookup过程 首先，**蓝色**结点确定**红色**结点负责保存key1。然后，**蓝色**结点将lookup或insert请求发送给**红色**结点。所以lookup过程的算法复杂度是O(1)。  ### 3.3 添加删除结点 当添加删除结点时，影响的只是很小的一部分数据。  ### 4 分布式哈希表 分布式哈希表DHT是一种概念模型或者说思想，其主要思路是：将每条文件索引K文件名或其他属性的哈希值-V存储文件的结点IP地址，组成一张巨大的文件索引哈希表。然后，再将这张大表按照一定规则分割成许多小块，并分布到各个结点上，每个结点负责维护其中一块。**DHT使用一致性哈希的思想来最小化拓扑结构变化带来的影响，并构造overlay网络实现高效地搜索**。 首先，将结点和数据映射到同一个线性地址空间，每个结点只负责地址空间中的一部分数据，但结点负责的信息通常是有重叠和冗余的。通常我们将这个线性地址空间看成一个环：  逻辑上地址空间形成的环对应着底层的物理拓扑结构，要注意的是真正的(underlay)的拓扑结构和逻辑的(overlay)拓扑结构通常是无关联的：  根据不同的DHT实现，查找过程会有不同。但不同的算法实现都类似于下图所示的过程，在地址空间中利用各种算法高效地找到负责保存数据的结点。注意，最后找到的数据分为两种： Ø  value就是我们要找的数据，它直接存储在结点上，这对于数据量不大的情况来说比较合适。 Ø  value不是我们要找的数据，而是数据存储在另一台机器的地址信息(如下图所示)。这种间接的存储方式多了一步数据的获取，但是对于大文件的存储来说更加合适。  ### 5 Chord实现 接下来要介绍的是最常见的MIT的Chord版DHT实现。 ### 4.1 Chord环和Finger表 首先，Chord使用SHA-1哈希函数(生成160位的id)和取模： Ø  **Node-id = SHA1(IP/mac)** Ø  **Key-id = SHA1(key)** Ø  **id-space mod ** ### 4.2 Lookup过程 对于Chord版的DHT实现来说，这种Lookup过程是通过一张叫做Finger表的路由表来完成的，它根据计算数据id指数级增长时对应的各个结点，形成表中的信息：  在没有finger表的情况下，需要不断访问后继结点继续lookup，即O(n)跳才能找到目标结点：  有了finger表，就可以实现O(logN)的高效lookup：  ### 5 算法复杂度对比 除了搜索/路由外，其他几项都是DHT占优：  ### 参考资料 1 Princeton - [P2P Systems and Distributed Hash Tables](http://www.cs.princeton.edu/courses/archive/spr11/cos461/docs/lec22-dhts.pdf) 2 Overlay Network：[http://en.wikipedia.org/wiki/Overlay_network](http://en.wikipedia.org/wiki/Overlay_network) 3 Peer-to-Peer：[http://en.wikipedia.org/wiki/Peer-to-peer](http://en.wikipedia.org/wiki/Peer-to-peer) 4 Structured Homogenous P2P Overlay Networks 5 [memcached全面剖析--4. memcached的分布式算法](http://blog.charlee.li/memcached-004/)

### 1 第三方缓存插件 除了Ehcache这种轻量级的缓存方案外，几乎所有IMDG产品都提供了对Hibernate二级缓存的直接支持，常用的有： Ø  Hazelcast Ø  GridGain Ø  JBoss Infinispan Ø  Terracotta(额外提供了直接替换Session对象的集成方式) ### 2 缓存工作过程 下面以JVM集群Terracotta为例，首先从最原始的JDBC到Hibernate到开启Hibernate二级缓存，看一下应用对数据库请求的情况。  ### 2.1 自动提交模式下的JDBC 在自动提交模式下使用JDBC时，JDBC不会对请求有任何缓存，每次SQL操作都会直接发送到数据库。因此下图中三个用户的访问会导致9次数据库访问。  ### 2.2 Hibernate一级缓存 Hibernate有两级缓存(参考各种Hibernate进行复习)，默认情况下第二级缓存是不开启的，后面会看到导致的问题。所以默认情况下，Hibernate会开启事务，并缓存更新操作，在最后事务提交时一起更新到数据库。算上commit事务提交的话，一共就是6次数据库访问。一级缓存在Session关闭时会自动清除，或者应用通过evict()清除某个对象、clear()清除全部缓存内容、flush()同步缓存与数据库。此外，与二级缓存相同的一点注意是：**使用HQL或SQL查询属性级别的数据时，是不会使用缓存的**。(第三部分缓存工作原理中会详细解释)  ### 2.3 集成Hibernate二级缓存 Hibernate的二级缓存进一步将多个Session的数据加载请求缓存。然而风险也随之而来，当开启二级缓存后Hibernate不再每次都请求数据库，于是缓存中的数据可能是过期的。有时我们可以使用TTL设置来强制Hibernate刷新，但有的场景下这种方案也是不可行的。此时，就可以使用Terracotta将这些二级缓存形成集群，Terracotta负责集群结点间数据的同步。这种方式只需改动Hibernate配置文件即可，但由于Hibernate对二级缓存的使用方式的限制，这种集成方式并不能最大限度地发挥Terracotta的威力。  ### 2.4 集成Hibernate Session Terracotta与Hibernate最快的集成方式就是直接与Hibernate的Session对象集成。因为POJO对象在Hibernate的二级缓存中实际是以字节数组的形式保存的，因此尽管已经在缓存中了，但也有序列化的开销。而使用这种集成Hibernate Session的方式，应用代码可以零延迟地直接访问内存中的POJO对象，没有任何多余开销。同样，我们也能够零延迟地在内存中写POJO对象。此时，数据库成了系统的历史记录，只在需要时更新。  ### 3 缓存内部工作原理 ### 3.1 二级缓存 之所以叫二级缓存是因为当你打开session时，Hibernate会自动为你开启一个一级缓存。官方文档中对二级缓存是这样描述的： A Hibernate Session is a transaction-level cache of persistent data. It is possible to configure a cluster or JVM-level (SessionFactory-level) cache on a class-by-class and collection-by-collection basis. You may even plug in a clustered cache. Be careful. Caches are never aware of changes made to the persistent store by another application (though they may be configured to regularly expire cached data). 像上面所提到的，只要SessionFactory开启着，二级缓存就存在。二级缓存持有每个标记为缓存的实体的所有**属性**和**关联**。下面以Person实体为例，说明一下二级缓存的内部工作原理：  我们先不开启关联的缓存，来看一下基本属性的缓存。二级缓存中，Person对象的保存格式如下：key是id，value是firstName，middleInitial，lastName三个String，外加parent的id。所以很明显，对象的实例没有直接保存在缓存中，而是被拆分成一个个属性保存，Hibernate将这个过程称为对象**脱水(dehydrate)**，反之属性组装成对象的过程称为补水(hydrate)。Hibernate为什么这样做？ Ø  用户代码对实体的实例上的修改不会破坏缓存，因为缓存里都是实例属性的拷贝。 Ø  对象间关联不容易过期，即使过期了也只是更新一个id就行了，因为缓存中不是直接保存parent变量对应的父Person实例，而只是其id。  不开启缓存和不开启关联缓存时，通过id加载操作对应的底层SQL执行情况如下。可以看出，如果不开启关联缓存的话，执行的SQL与不开启缓存时几乎一样(只省掉了查询父Person的第一条SQL)：  现在开启关联缓存来看一下其效果。开启关联缓存后，Person的缓存格式如下。缓存内容多了子Person对象的id集合，与父Person缓存类似，也是只保存id而不是所有子Person实例。至此，**Person对象本身以及嵌套的parent(Person)和children(Set)都被完全脱水成属性和关联id**：  此时，**对Person对象的通过id加载操作完全不需要访问数据库了**，所以一定要开启关联缓存才能真正发挥出二级缓存的效果。但是，当执行HQL或SQL查询时，依然不会使用二级缓存。例如，我们通过firstName进行HQL查询时，Hibernate会先执行一条具有相同where条件的SQL获取出Person的id，有了id之后才能开始使用二级缓存中的内容，也就是说：**二级缓存只能在通过id查询时有用，对于其他复杂的条件查询是失效的**。而这条**查询出id的相同where条件的SQL就是查询缓存能发挥用处的地方**，于是就引出了查询缓存。  ### 3.2 查询缓存 首先要配置开启查询缓存 true 此外还要在使用Query对象前调用setCacheable(true)。 查询缓存也是key-value的形式，其key是HQL或SQL以及参数，而value是查询结果的id集合，所以其value与开启关联缓存后的二级缓存非常像。所以，**只有在HQL或SQL一致，并且查询参数也完全一致的情况下，查询缓存才会有用！**现在来看一下两种缓存共存时的效果。当二级缓存的关联缓存和查询缓存都开启时，**当我们执行一个特定参数的HQL/SQL，首先会去查询缓存中找到对应的id，再去二级缓存中找到这些对象，以及这些对象关联的对象集合**。  ### 3.3 总结 不论是二级缓存还是查询缓存，本质上都是key-value形式保存的，所以**只能对key(id或HQL/SQL+参数)进行匹配查询。从这点上看，Hibernate缓存不够灵活，不能像IMDG产品那样，真正的将对象缓存在内存中，并提供面向对象的查询**。理解了Hibernate缓存的底层工作原理，才能更好的管理我们的缓存数据，理解Hibernate缓存的行为。 ### 参考资料 1 《The Definitive Guide to Terracotta》 2 [Hibernate: Truly Understanding the Second-Level and Query Caches](http://www.javalobby.org/java/forums/t48846.html)

### 陷阱 使用cache API时，一个最重要的问题就是潜在的数据加载。因为IMDG提供的分布式集合也都是实现的JDK的Map、Set等接口，以JDK的Map为例，它接口规定put和remove返回被替换的对象或删除掉的对象，所以这会导致我们操作缓存时导致与后端存储的通信。所以我们要调用其它版本的API，例如GridGain提供了putx等接口，其返回值是一个布尔值而非旧的对象。 ### 问题 现今的IMDG虽然已经比较成熟，但仍有一些问题需要解决： Ø  冷启动问题：即便是能够并行还原，从后端数据库还原数TB的数据依然需要很久，这段时间是会导致数据库的过载，可能无法为其他应用提供服务。 Ø  缺少CDC能力：还没有一款IMDG产品提供CDC(Change Data Capture)能力，即数据库数据的变化如何反映到IMDG中。现在一般都需要触发器或第三方产品来实现CDC，例如Hazelcast通过第三方Speedment能够监控数据库中的数据变化。 Ø  分布式计算容错：像Impala分布式计算平台直接不提供容错。 ### 参考资料 1 [Hazelcast's MapLoader Pitfalls](http://java.dzone.com/articles/hazelcasts-maploader-pitfalls) 2 [Jags Ramnarayan on In-Memory Data Grids](http://www.infoq.com/articles/in-memory-data-grids)

### 1 简介 将内存作为首要存储介质不是什么新鲜事儿，我们身边有很多主存数据库(IMDB或MMDB)的例子。在对主存的使用上，内存数据网格(In Memory Data Grid，IMDG)与IMDB类似，但二者在架构上完全不同。IMDG特性可以总结为以下几点： Ø  数据是分布式存储在多台服务器上的。 Ø  每台服务器都是active模式。 Ø  数据模型通常是面向对象和非关系型的。 Ø  根据需要，经常会增减服务器。 此外，IMDG与普通缓存系统也是不同的。同样地，在主存使用以及水平扩展上缓存系统与IMDG类似。但是，两者的使用方法和目的是完全不同的。缓存系统只是缓冲读压力，像RDBMS这种持久化存储是必备的。例如下图中的Arcus缓存系统。而IMDG的架构请参考第二部分。  换言之，**IMDG将对象本身存储在内存中，并保证可扩展性**。常见的商业以及开源产品如下： Ø  Hazelcast Ø  Terracotta Enterprise Suite Ø  VMware Gemfire Ø  Oracle Coherence Ø  Gigaspaces XAP Elastic Caching Edition Ø  IBM eXtreme Scale Ø  JBoss Infinispan ### 2 架构 IMDG亟需克服的两个核心问题是：容量限制和可靠性。通常，IMDG通过水平扩展来克服内存容量上的限制，而通过复制系统来保证可靠性。典型的IMDG架构如下图所示。  因此，前面介绍过的缓存系统与IMDG的区别很明显。  ### 3 特性 除了提供各种数据结构的分布式实现外，IMDG一般会使用堆外内存(off-heap，或叫弹性内存)来降低垃圾回收的压力。  ### 参考资料 1 [Introduction to In-Memory Data Grid: Main Features](http://www.cubrid.org/blog/dev-platform/introduction-to-in-memory-data-grid-main-features/)

### [1 内存计算与云计算]() 如果说云计算这个新瓶装的是[虚拟化+ SOA/网格计算+SaaS(软件即服务)](http://soft.chinabyte.com/168/11106668.shtml)的老酒，那么内存计算则重点是释放了计算这一部分的能量。  但是对内存计算经常有一些[误解](http://www.tuicool.com/articles/VJNFf2)： Ø  **大容量内存很贵** Ø  **内存计算不会持久化**：实际上几乎所有的内存计算中间件都提供多种内存备份、持久存储备份以及基于磁盘的swap空间溢出的策略。 Ø  **内存计算要取代数据仓库**：内存计算的目的是要改善那些需要OLTP和OLAP混合处理的可操作数据集(Operational Dataset)的计算，而非历史数据集(Historical dataset)。简言之，内存计算不是要把企业的所有数据都放进内存。 Ø  **闪存已经足够快了**：内存计算不是要达到2-3倍的边际效应提升(Marginal Effect)，而是10-100倍的提升，使之前那些不可行的业务和服务成为可能。 Ø  **内存计算等于内存数据库**：首先，内存计算是一种技术而不是某种产品。其次，内存数据库只是目前内存计算触手可及的成果，内存计算长期的发展还是在流式处理(Stream Processing)上。此外，内存计算与传统内存数据库的区别是：内存计算是为分布式、弹性环境以及内存数据处理而设计的。 ### 2 内存计算产品分类 根据内存计算技术的发展顺序，内存计算大致可以分为[三类产品](https://gridgaintech.wordpress.com/2013/11/19/cache-data-grid-database/)： Ø 分布式缓存(Memcached/Redis)：主要使用场景就是将频繁访问的数据保存在内存中避免磁盘加载。多数产品都是分布式**内存key/value存储，并提供简单的put和get方法**。随着不断成熟，与后端的read/write-through，ACID事务，复制和分区，eviction策略等也逐渐加入到产品中，这些特性也成为了后来出现的IMDG/IMCG产品的基础。 Ø  内存数据/计算网格(IMDG/IMCG, GemFire/Hazelcast/GridGain)：数据网格的显著特性是**co-location计算**，将计算过程发送到数据本地执行。这是数据/计算网格的关键创新点，在数据量不断增长的情况下再加数据抓取过来执行计算已经变得不现实了。这种创新也不仅使内存计算从简单的缓存产品进化，也激发了后来IMDB的诞生。 Ø  分布式内存数据库(IMDB, VoltDB/Impala)：分布式内存数据库的显著特性是增加了**基于标准SQL或MapReduce的MPP(大规模并行处理)能力**。如果说数据网格的核心是解决数据量不断增长下计算的困境，那么分布式内存数据库就是解决计算复杂度不断增长的困境。它提供了分布式SQL、复杂(分布式共享)索引、MapReduce处理等工具。  值得注意的是，随着技术的发展，有些界限不再那样清晰。像现在很多IMDG产品已经具有IMDB的特性，能够提供复杂的分布式SQL和MapReduce计算能力。不管怎样，这些产品中的核心技术和算法都是不变的，所以学习时不必过分纠结某个产品到底属于那一类。  ### 3 应用场景 以上提到的各种内存计算产品可以应用到大数据处理的各个环节上，如下图所示，其中涉及内存计算的技术标成红色。  [ ]() 1）事务处理：主要分为Cache(Memcached, Redis, GemFire)、RDBMS、NewSQL(以VoltDB为首的)三部分，缓存和NewSQL数据库是关注的重点。 2）流式处理：Storm本身只是计算的框架，而Spark-Streaming才实现了内存计算式的流处理。 3）分析阶段的对比： Ø  通用处理：MapReduce，Spark Ø  查询：Hive，Pig，Spark-Shark Ø  数据挖掘：Mahout，Spark-MLLib，Spark-GraphX 从上可以看出，Spark生态圈的子项目以及Impala都是值得关注的重点。 ### [4 核心技术]() [因为内存计算主要释放了云计算中的计算部分的能量，所以它主要涉及**并行/分布式计算**和**内存数据管理**这两大方面的技术体系：]() Ø  并行/分布式计算：网络拓扑、RPC通信、系统同步、持久化、日志。 Ø  内存数据管理：字典编码、数据压缩、内存中数据格式、数据操作、内存索引、内存中并发控制和事务。 以下简单列举了内存计算主流产品中涉及到的技术点： Ø  Memcached/Redis：一致性哈希。 Ø  GridGain：DHT、refresh-ahead、off-heap、continuous query。 Ø  Infinispan：LIRS eviction。 Ø  Spark(LMAX)：immutable model(RDD)。 Ø  VoltDB：single-threaded。 Ø  Impala(Dremel)：LLVM optimizing、nested record、MPP。 ### 理论 **直接在Google或Bing.com搜索"论文名 pdf"** 《Implementation Techniques for Main Memory Database Systems》 《Main Memory Database Systems: An Overview》 《The Revolution in Database Architecture》 – Jim Gary 《A Study of Index Structures for Main Memory Database Management Systems》 《High-Performance Concurrency Control Mechanisms for Main-Memory Databases》 《A bridging model for parallel computation》 (BSP model) 《SEDA: An Architecture for Scalable, Well-Conditioned Internet Services》 [实时处理与流处理](http://blog.csdn.net/dc_726/article/list/2) [关系代数的并行计算](http://blog.csdn.net/dc_726/article/details/41909647) [并发计算模型BSP与SEDA](http://blog.csdn.net/dc_726/article/details/41628413) [从NSM到Parquet：存储结构的衍化](http://blog.csdn.net/dc_726/article/details/41777661) ### 产品 **GemFire，VoltDB：** [分布式缓存GemFire架构介绍](http://blog.csdn.net/dc_726/article/details/41378633) [NewSQL数据库VoltDB特性简介](http://blog.csdn.net/dc_726/article/details/41909719) **Spark：** [Spark分布式计算和RDD模型研究](http://blog.csdn.net/dc_726/article/details/41381791) [Spark发展现状与战线](http://blog.csdn.net/dc_726/article/details/41552421) [分布式内存文件系统Tachyon](http://blog.csdn.net/dc_726/article/details/41552593) **Impala(Dremel)：** [Google Dremel数据模型详解(上)](http://blog.csdn.net/dc_726/article/details/41627613) [Google Dremel数据模型详解(下)](http://blog.csdn.net/dc_726/article/details/41777619) [Impala中的代码生成技术](http://blog.csdn.net/dc_726/article/details/41778611)

VoltDB是一个革命性的新型数据库产品，被称作NewSQL数据库。它基于H-Store，号称比当前数据库产品的吞吐量高45倍，同时又具有很高的扩展性。它的特性主要有以下几点： Ø  高吞吐、低延迟：通过内存计算，存储过程和串行数据访问实现。 Ø  可扩展性：自动分区和复制，保证性能和可扩展性。 Ø  高可用性：同步的多主复制(在VoltDB中叫K-safety)。 Ø  持久化：数据库快照与命令日志(command log)的创新技术组合。 ### 1 高吞吐、低延迟 VoltDB能够提供高吞吐、低延迟的SQL操作，总体来说，它是通过内存计算避免磁盘阻塞(disk stall)，通过存储过程避免用户阻塞(user stall)，通过集群结点内的数据访问串行化，避免传统数据库锁、缓冲管理的开销。此外，VoltDB并不是纯Java开发，其SQL执行引擎是C++写成的，所以并不受GC暂停的影响。 Ø  **内存计算**：使VoltDB在事务执行期间无需等待磁盘加载，避免磁盘I/O开销。充分利用了现代服务器上庞大的内存，将吞吐量最大化。 Ø  **存储过程**：避免应用与数据库之间的多次通信开销，每个事务被定义成一个存储过程，因此事务只需一次通信往返。然而，VoltDB并不是只支持存储过程，从1.1版本开始已经能够支持来自JDBC、SQL命令行、HTTP/JSON、原生C++/PHP/C#/Node.js等等客户端的SQL查询。唯一的限制就是：**VoltDB总是自动提交模式，不支持手动控制事务**。 Ø  **数据访问串行化**：传统数据库在前面两种阻塞等待的情况下，会切换执行其他事务，因此会导致很大的锁(latching and locking)开销。而一个VoltDB数据库由许多内存计算引擎组成(叫做partition分区)，每个分区都是数据和相关处理过程的集合。VoltDB在集群内自动分发数据创建分区，每个分区内都是单线程的，从而避免了传统数据库对并发控制的开销。 Ø  **C++执行引擎**：VoltDB使用原生C++代码进行表数据的内存分配和SQL的执行，之所以核心不使用Java就是避免将表数据这种长时间存活的数据放置到JVM堆上，同时对内存使用进行更细粒度的控制。此外，像静态的部署相关和schema相关的数据，尽管是在Java中管理，但也使用DirectByteBuffer分配到堆外内存。所以其实JVM堆只是用来分配事务相关的一些存活期很短的数据，这对于GC来说是合适的负载。 如果某个事务只涉及一个单一分区内的数据，则其处理流程如下图所示：  ### 2 扩展性架构 从架构上看，VoltDB属于shared nothing架构，因此可以很容易地实现扩展，可以通过增加已存在结点的容量和性能实现垂直扩展，通过动态增加新结点实现水平扩展，而在这个过程中不需要修改任何数据库schema和应用程序代码。 同时，VoltDB不仅支持表分区，还支持表复制。对于大表，可以通过分区来提高性能。对于频繁读取的小表，可以通过复制来减少join。  这与分布式缓存GemFire中的mirrored region和partitioned region的概念很像。在GemFire这，mirrored region包含全量数据，而partitioned region只包含分区数据。但不同的是，VoltDB是根据表的特点选择复制或分区，而GemFire则通过mirrored region将其他分区数据抓取到一起形成全量的数据镜像。  如果一个事务涉及多个分区的数据访问，那么其处理流程如下图所示。一个结点会充当协调者(coordinator)，负责分发任务给其他结点，并收集结果，完成任务。  ### 3 高可用性 不像传统RDBMS产品依赖第三方的HA解决方案，VoltDB提供三种HA能力：K-safety，网络故障检测，存活结点重连(rejoin)。 ### 3.1 K-safety 当配置成K-safety时，VoltDB会自动地复制数据库分区，K表示副本的个数。例如K=0时表示没有副本，所以任何一个结点的故障都会导致整个数据库集群的停止服务。当K=1时表示有1个副本，即一共2份拷贝。要注意的是：VoltDB中的副本是可以读写的，而不是传统的主从复制关系。  关于数据同步问题的解决，**任何发生在复制分区上的操作都会发送给各个拷贝的结点去执行**，来保证一致性。如果其中一个结点失败，那么数据库会继续发送这个操作给失败的结点。因此在这一点上VoltDB与传统数据库有很大不同，不存在多主(multi-master)情况下的数据同步冲突问题。所以K-safety也叫做同步多主复制。 ### 3.2 网络故障检测 当网络发生故障时，VoltDB的结点彼此之间被物理隔离开，而认为对方已经发生故障。那么K-safety机制会使这两侧的结点继续分别提供服务。如果不及时检测到的话，这种“分离的大脑”(split brain)会导致严重的数据同步问题。因此，VoltDB会自动检测网络故障，立即评估出那一侧结点应该继续服务，并快照另一侧的结点数据后停掉服务。当网络故障解决时，可以直接使用下面将介绍到的存活结点重连技术将结点重新加入到集群中。  ### 3.3 存活结点重连 离线的VoltDB结点可以通过rejoin操作重新加入到集群中。具体过程是：首先从兄弟结点获得一份数据拷贝，当追赶上兄弟结点时，此存活结点就可以回到正常状态，接受任务了。 ### 4 持久化 尽管VoltDB的HA能够降低当机概率，但故障还是偶尔会发生，而且DBA有时也要定期地停机维护。因此，VoltDB提供了高性能的快照和命令日志(command log)来支持各种持久化需求。对于日志，VoltDB支持同步和异步，以及刷新到磁盘的时间间隔等配置。  那command log与传统的WAL(write-ahead log)有什么区别呢？(待深入研究)  ### 总结 但这样也不代表VoltDB是万能的，其设计和特性决定了其应用场景，VoltDB比较适合高频率请求、短事务的应用，像金融、零售、Web2.0等，以及流式数据应用，像推荐引擎、实时广告平台、点击流处理、欺诈交易检测等。 ### 参考资料 1 VoltDB Technical Overview 2 Using VoltDB 3 [Debunking Myths about the VoltDB in-memory database](http://voltdb.com/blog/debunking-myths-about-voltdb-memory-database) 4 [Impact of Java Garbage Collection on in-memory databases](http://voltdb.com/blog/impact-java-garbage-collection-memory-databases) 5 [Command logging vs. Write-ahead Logging](http://stackoverflow.com/questions/14181180/why-do-sql-databases-use-a-write-ahead-log-over-a-command-log)

Cloudera Impala是一种为Hadoop生态系统打造的开源MPP(massive parallel processing)数据库，它主要为分析型查询负载而设计，而非OLTP。Impala能最大限度地利用现代硬件和高效查询执行的最新技术。LLVM下的运行时代码生成就是用来提升执行性能的技术之一。 ### LLVM简介 LLVM是一个编译器及相关工具的库(toolchain)，它不同于独立应用式(stand-alone)的传统编译器，LLVM是模块化且可重用的。它允许Impala这样的应用在运行的进程内执行JIT(just-in-time)编译。尽管LLVM因一些特殊的能力以及著名的工具，如比GCC更优的Clang编译器，但真正使LLVM区别于其他编译器的是它的内部架构。 经典静态编译器(像多数C编译器)中，最流行的设计是前端、优化器、后端组成的三阶段设计。前端解析源码并生成抽象语法树(AST，Abstract Syntax Tree)。优化器会做很多优化来提升代码性能。后端(或称代码生成器)将代码转换成目标平台的指令集。这种模型对解释器、JIT编译器。JVM也是这种模型的一种实现，它使用字节码作为前端和优化器之间的接口。  这种经典设计对于多语言(包括源语言和目标语言)支持非常重要。只要优化器内部使用一种公共代码表示，前端和后端就能够编译任意的语言。当需要移植(porting)编译器来支持一种新语言时，只需实现一个新前端，而优化器和后端都可以重用。否则就要重新实现整个编译器，支持M种源语言*N种目标语言。  尽管各种编译器教科书中都讲到三阶段设计的种种优点，但实际上它从未被实现过。像Perl、Python、Ruby和Java的编译器实现并没有共享任何代码。此外，还有各种各样的特殊用途的编译器，例如图像处理、正则表达式等CPU密集型的子领域的JIT编译器。GCC由于混乱的代码结构而无法提取出可重用的组件，例如前端和后端重用了某些全局变量等，所以我们无法将GCC嵌入到应用程序中。下图是LLVM对三阶段设计的实现。  ### Impala中的LLVM Impala使用LLVM在运行时产生完全优化并且查询特定的函数，这比通用的预编译函数有更好的性能。**尤其是会在一次查询中执行许多次的内层循环(inner loop)的函数**。例如，一个用来解析文件记录并装载进Impala内存元组的函数，在每个文件的每一条记录被扫描时都会被调用。对于这种函数，即使只是简单的移除一些指令也会得到速度上的巨大提升。 如果没有运行时的代码生成，为了处理编译时未知的运行时数据，函数中总是会包含低效的代码。例如，仅仅处理整数的记录解析函数，在处理只有整数的文件时，会比处理各种数据类型的通用函数要快得多。然而要扫描的文件schema在编译时是未知的，所以这种通用的函数尽管低效，却也是必要的。 下图1中的代码示例。编译时记录个数和类型都是未知的，所以**处理函数要写的尽可能通用，避免发生未考虑到的情况。但JIT与这种思路完全相反**，函数在运行时被完全编译成对当前数据最高效的写法。这在我们平时看来甚至都不能算作函数，因为完全不通用，逻辑都用常量固定写死了，但这正是JIT的策略！所以像下面的动态生成的MaterializeTuple对于不同的运行时信息(如不同的查询)会有完全不同的生成版本。  代码生成中的常用优化技术： Ø  **移除条件分支**：因为已知运行时信息，所以可以优化if/switch语句。这是最简单有效的方式，因为最终机器码中的分支指令会阻止指令的管道化(instruction pipelining)和并行执行(instruction-level parallelism)。同时，通过展开for循环(因为我们已经知道循环次数)和解析数据类型，分支指令能被一起移除。 Ø  **移除内存加载**：从内存加载数据是开销很大而且阻止管道化的操作。如果每次加载的结果都一样的话，我们就可以使用代码生成来替代数据加载。例如，之前图1中的数组offsets_和types_在每次查询开始时创建而不会改变，于是在代码生成的函数版本中，展开for循环后，这些数组中的值可以直接内联化。 Ø  **内联虚函数调用**：虚函数对性能的影响很大，尤其是函数很小很简单，因为它无法内联化。因此当对象实例的类型在运行时可知时，我们可以使用代码生成来取代虚函数的调用，并做内联化。这对于表达式树的求值尤为有价值。在Impala中，表达式由操作和函数的树组成，例如下图2。树中出现的每种表达式都是覆盖(override)表达式基类的函数来实现的，基类会递归地调用各个子表达式。许多表达式函数都是非常简单的，例如两数相加，于是**虚函数调用的开销甚至大过表达式求值的开销**。通过代码生成移除虚函数并内联化，表达式可以无需函数调用而直接求值。此外，内联后的函数使编译器做进一步的优化，例如子表达式消除等。  ### 用LLVM生成代码 当Impala受到查询计划(query plan，由Impala的Java前端负责生成)时，LLVM会被用来**在查询执行开始前**，生成并编译对性能至关重要的函数的查询特定版本。LLVM主要使用IR(intermediate representation)来生成代码，例如LLVM的前端Clang C++编译器生成IR，LLVM优化IR并将其编译成机器码。IR类似于汇编语言，由一些简单的、能够直接映射成机器码的指令组成。在Impala中有两种技术来生成IR函数：使用LLVM的IRBuilder API来编程式地生成IR指令；使用CLang将C++函数交叉编译成IR。 下图是IR的例子。可以看出，IR是一种类RISC的虚拟指令集。它支持加减、比较、分支等指令。此外，IR还支持标签。但与多数RISC不同的是： Ø  LLVM是强类型的，它有一套简单的类型系统，例如i32, i32**，add i32。 Ø  LLVM IR支持无限的临时寄存器，以%开头。 **因为优化器不受源语言和目标平台限制，所以IR的设计也要遵守这个原则。**  在LLVM中，优化器被组织成优化pass的管道，常见的pass有内联化、表达式重组、循环不变量移动等。每个pass都作为继承Pass类的C++类，并定义在一个私有的匿名namespace中，同时提供一个让外界获得到pass的函数。  我们可以决定pass的执行顺序甚至是否执行。当我们实现一种图像处理语言的JIT编译器时，我们可以去掉没用的pass。例如，如果通常都是大函数的话，就没必要浪费时间内联。如果指针很少的话，那么别名分析和内存优化就变得可有可无。但是LLVM不是万能的，PassManager本身也并不知道每个pass内部的逻辑，所以这还是由我们实现者来确定的。  ### 参考资料 1 Runtime Code Generation in Cloudera Impala 2 The Architecture of Open Source Application [http://www.aosabook.org/en/llvm.html](http://www.aosabook.org/en/llvm.html)

###  “神秘”的r和d **单从数据结构来看的话**，我们可以这样解释r和d的含义。r代表着当前字段与前一字段的关系，是在哪一层合并的，即公共的父结点在哪？举例来说，假如我们重建到了Code='en'，通过r=2可以知道是在Language那一层发生了重复。  为了保持原纪录的结构，我们会保存一些NULL数据，而d就是用于重建NULL字段。通过d的值，就能知道NULL的结构。例如下图，通过r=1知道应该合并到Name那一层。而通过d=1则知道路径上只有一个字段，即不仅仅是Code字段不存在，Language也不存在。这样就把NULL正确地重建出来了，那么接下来的Code='en-gb'的层级也就不会乱了。  然而这只是从静态的数据结构来解释，而r和d的深层次含义还是要看FSM是如何执行的。**真正的因果关系是FSM的执行方式决定了数据结构的设计**。 ### 3 记录查询 ### 3.1 从FSM角度看r和d 先看一下前面例子的完整FSM的样子。如果把Protocol Buffer中对数据格式定义的schema看作是编译原理中的语法定义的话，那么一般可以使用工具如antlr, yacc自动生成自动机，手写的话是相当恐怖的吧。  对列数据的完整遍历就是这个样子的：  在讨论查询如何执行之前，先继续刚才未完成的题目，r和d的本质，这次通过动态的FSM的角度来分析，而不是静态的数据结构了： Ø  **FSM状态机只是定义了状态的变更，即处理流程应当如何在各个列的存储表之间跳转，而实际数据还是在表中保存**。有点像数据库索引，遍历时是根据FSM进行跳转，然后对某一列的表进行table scan。但索引是靠字段值的顺序组织，因为数据库表之间没什么嵌套关系，而Dremel的FSM则是靠字段之间的嵌套关系来组织。  Ø  状态机中线条上的数字表示什么？回忆一下，数字表示的是：字段的数据表中当前行的下一行的r值。通过检测下一个r值来决定跳转。因为**r=0，则说明下一行与当前行所表示的字段一定不在同一路径，否则必然会在某一Level上有共同的字段(路径的部分重叠)**。注意这是由于Protocol Buffer的schema不是树，没有共同的根所导致，否则所有字段必然都会在根重复，上面对r的解释也就没意义了。以repeated的Forward为例，检查到下一行r=1说明40、60都是接在20字段下面的。Code字段也是同样道理。  Ø  Name.Language.Code到Name.Language.Country之间的线上为什么是0，1，2？因为Name.Language.Code是required不是repeated，读取后不管下一行的r值是多少都要去读Name.Language.Country。同理Name.Language.Country也是读完不管怎样都跳到下一字段。 Ø  最复杂的要属Name.Url了，因为它是schema里定义的最后一个字段。在Name.Url这要决定到底是继续下一文档如r2的处理，还是跳回到本文档的其他字段继续处理。具体分析一下：**r=0说明当前文档中没有Name字段了**。为什么这么说？因为如果文档后面真有Name字段，假如下面有Url，则当前表中的下一条应该是r=1；**假如下面没有Url，则当前表的下一条应该是r=0的NULL。这里NULL又发挥用处了！所以中间部分的NULL能保持结构无损，而后面部分的NULL能提示文档是否结束**。 ### 3.2 查询引擎 至此，我们已经彻底摸清Dremel数据模型以及FSM的基本运行方式了。现在终于可以分析Dremel是如何解析和执行类SQL查询的了。查询语言类似SQL，输出也是个嵌套式的记录，以及schema定义。  那么查询引擎如何执行呢？首先为查询语句中涉及到的每个字段都打开一个Reader来读取数据，然后就是根据WHERE中的条件过滤以及根据SELECT中的条件投影并聚合了。难点在于：**重建出层次关系，再进行过滤和聚合**。例如，过滤掉DocId=20很容易，但其实文档r2的所有记录都应被过滤。因为WHERE中两个条件是AND关系，同时DocId又是最底层的字段，所以相当于r2这一整棵树都被裁剪掉了。Code=en-gb也是由于所在的Name字段下没有满足http开头的Url字段，而被间接的过滤掉了。 聚合也是同样道理，有了层次关系，才能正确的聚合。例如Code=en-us，en和Url=http://A是同一个Name下的，COUNT和字符串拼接时会一起处理。而Url=http://B则是另一个Name下的，要分开处理。  ### 参考资料 1 Dremel: Interactive Analysis of Web-Scale DataSets

首先简单介绍一下Dremel是什么，能解决什么问题。第二部分着重讲Dremel的数据模型，即数据结构。第三部分将谈一下在此数据结构上设计的算法。 ### 1 起源 Dremel的数据模型起源于分布式系统的应用环境(Protocol Buffers，一种在Google内广泛使用，现已开源的实现)。其数据模型是基于强类型的嵌套记录，抽象语法可以表示成下面公式：  一个例子：  ### 2 嵌套列式存储 ### 2.1 记录结构的无损表示 首先来看一下Dremel的数据模型是如何在列式存储下无损的表示出记录的结构的(lossless representation of record structure in a columnar format)。如果仅仅是数值(values)的话，数值本身无法传递出记录(record)的结构信息。我们不知道两个数值是属于两条不同的记录还是在一条记录下，同时我们也不知道一些可选的字段(field)是否显式定义。因此，我们引入了两个概念：**Repetition Level**和**Definition Level**。 为了说清楚Dremel模型是如何无损地表示数据的，我想到了两种画法。最终还是决定采用第一种画法，类似有向图，感觉与后面的FSM状态机能更好的对应上。  ### Repetition Level Dremel论文中对repetition level的定义听起来比较抽象：***at what repeated field in the field's path the value has repeated***。意思就是在路径上，在哪个repeated字段上重复了。还是看个例子解释一下吧，以之前的图例中的文档r1中的Code字段为例。  上图清晰地表示出三个Code字段与文档中字段的对应关系。下面来看一下这三个Code的repetition level(简写为r) 0，2，1是如何计算出来的。下图忽略无关的字段，将三个Code字段的完整路径都表示出来。那么就可以简单易懂地看出，r就是这些字段路径上，发生重复了的字段的level。请参考下图中的注释就能很快理解。  大家可能还注意到Name.Code表中除了en-us、en和en-gb三行外，还有两行NULL。第二个NULL是描述文档r2的，我们就分析一下第一个NULL的含义吧。因为文档r1的第二个Name字段下没有Code，而为了说明en-gb是属于第三个Name字段下的，所以在en和en-gb之间加了一行NULL，其r也等于1(Name重复)。同时，由于Code在定义中是required的字段，所以事实上这一行NULL也暗示了：在第二个Name字段下Language也是不存在的。不然Language存在而下面却没有Name，这是不符合文档定义的。 以此类推，其他字段的r值都是这样计算出来的。同时注意一点：我们只保存了有值的字段，如DocId、Name.Url、Name.Language.Code等，而像Links、Name.Language等字段是没必要保存的。  ### Definition Level definition level(简写为d)在论文中的定义还比较清楚：Each value of a field with path p , ***esp. every NULL***, has a definition level specifying ***how many fields in p that could be undefined***(because they are optional or repeated) ***are actually present***. 尤其对于NULL来说，路径p上有多少字段可以是不存在(例如在文档定义中是optional或repeated，而不是required)，然而实际却存在的。例如文档r1的Links下没有Backward字段，然而Links字段却存在(因为Links下有Forward)，所以我们在Links.Backward表中保存一条NULL，并且d=1。对于非NULL字段来说，意义不大，因为d的值对于每种字段来说都是相同的，例如Code都是2，Country都是3。  值得注意的几点是： Ø  在路径上计算多少字段本可以不存在时，包含了当前字段本身。例如计算Country:us时，Country本身也是optional，也计入总数，所以d=3。 Ø  每种字段只计算1次。例如最下面的Country:gb，在其路径上的3个Name都满足条件，但只计1次，所以d=3，而不是5。(前面提过，也许是我这第一种画法的缘故，需要这一条规则来限定) ### 数据压缩 前面介绍了数据的保存方法，实际上真正保存时，数据还会被进一步压缩。 Ø  不显式保存NULL，因为它可以通过d来确定：d < 路径上repeated和optional字段总数，就说明是NULL。可以通过前面的例子印证一下。 Ø  总是会被定义的字段的d不会被保存。 Ø  r也是仅在必要时才会保存。例如d=0暗示r=0，所以r可以省略不存。 Ø  像DocId这种所有level都是0的，实际上不会保存任何level信息。 Ø  尽可能使用位图。例如假如d最大是3，那么我们只使用2个bit来保存。 ### 2.2 快速编码成列式存储 略，详见论文附录部分的伪代码。 ### 2.3 高效地组装记录 高效地从列式存储数据中组装出记录，对像MapReduce这种面向记录的数据处理工具来说非常重要。我们的目标是：给定字段的子集，我们能重新构建出仅包含选中字段的原始记录，而过滤掉其他字段。**核心思想是：使用有限状态机(finite state machine, FSM)**读取每个字段的值和level，顺序地追加到输出流中。FSM为每种字段都关联一个field reader。状态转变通过repetition level来标记。一旦reader抓取到值，我们继续看下一repetition level来决定使用哪个reader。FSM就这样从开始状态到结束状态遍历完每条记录。 下面还是用前面的例子，通过DocId和Name.Language.Country这两个字段的重建，来详细解析一下FSM的工作过程。关键步骤用红色加粗标记。  1.      FSM委托Reader1读取DocId第一行，通过r=0重建记录。 2.      检查DocId第二行，发现r=0，则**Reader1停在当前“游标”位置**。FSM将状态变化到Name.Language.Country。 3.      FSM委托Reader2读取Name.Language.Country第一行，通过r=0重建记录。 4.      FSM委托Reader2读取Name.Language.Country第二行。**通过r=2(说明Language字段重复，即Language有多个)重建记录。** 5.      FSM委托Reader2读取Name.Language.Country第三行。**通过r=1和d=1(说明只有Name字段不是NULL)重建记录。** 6.      略过第四行。 7.      检查到第五行，发现r=0，Reader2停在当前位置。FSM再次发生状态变化，继续重建文档2的记录。 8.      FSM委托**Reader1继续读取DocId第二行(之前Reader1就停在这里了)**。 9.      到这里应该已经很清楚了，最后过程就略说了：DocId中没有数据了，FSM状态变化，Reader2继续读取Country的最后一行数据，重建出记录。 *注：论文原图中少了第二个Name字段，我觉得应该加上吧。在第五步被重新构建出来。为什么在原图中没有呢？* 前面例子的完整FSM就是这样的： 

### 前言 现今Spark正是风头正劲时，Spark本是UCBerkeley的AMPLab诞生的项目，后来捐赠给了Apache来管理源码和后续发展。今年从Apache孵化器终于孵化出了1.0版本。其对大数据的支持从内存计算和流处理，到交互式查询，一直到图计算和机器学习，可谓摆开了架势、拉长了战线，一方面挑战老前辈Hadoop和MapReduce，另一方面又随时准备迎接同样的后起之秀的挑战。 ### 大数据的今天 今天的大数据系统生物圈百花齐放，有已经如日中天的通用批处理MapReduce，也有针对不同应用场景而特殊化的处理系统。  ### 全栈式的Spark Spark作为后起之秀，以其RDD模型的强大表现能力，不断完善自己的功能，逐渐形成了一套自己的生物圈，提供了full-stack的解决方案。其中主要包括Spark内存中批处理，Shark交互式查询，Spark Streaming流式计算三大部分。此外还有GraphX和MLBase提供的常用图计算和机器学习算法。  而Spark由于采用Scala编写，底层使用Akka，代码十分简洁。而且借助RDD的强大表现力，Spark各种子项目的代码量也很小。  ### Spark使用情况 援引自[一篇博文](http://www.beagledata.com/news/764.html)，看一下Spark在互联网界的使用情况。 **1. 腾讯** **“**广点通是最早使用Spark的应用之一。腾讯大数据精准推荐借助Spark快速迭代的优势，围绕“数据+算法+系统”这套技术方案，实现了在“数据实时采集、算法实时训练、系统实时预测”的全流程实时并行高维算法，最终成功应用于广点通pCTR投放系统上，支持每天上百亿的请求量。 基于日志数据的快速查询系统业务构建于Spark之上的Shark，利用其快速查询以及内存表等优势，承担了日志数据的即席查询工作。在性能方面，普遍比Hive高2-10倍，如果使用内存表的功能，性能将会比Hive快百倍。 **2. Yahoo** **“**Yahoo将Spark用在Audience Expansion中的应用。Audience Expansion是广告中寻找目标用户的一种方法：首先广告者提供一些观看了广告并且购买产品的样本客户，据此进行学习，寻找更多可能转化的用户，对他们定向广告。Yahoo采用的算法是logistic regression。同时由于有些SQL负载需要更高的服务质量，又加入了专门跑Shark的大内存集群，用于取代商业BI/OLAP工具，承担报表/仪表盘和交互式/即席查询，同时与桌面BI工具对接。目前在Yahoo部署的Spark集群有112台节点，9.2TB内存。 **3. 淘宝** **“**阿里搜索和广告业务，最初使用Mahout或者自己写的MR来解决复杂的机器学习，导致效率低而且代码不易维护。淘宝技术团队使用了Spark来解决多次迭代的机器学习算法、高计算复杂度的算法等。将Spark运用于淘宝的推荐相关算法上,同时还利用Graphx解决了许多生产问题，包括以下计算场景：基于度分布的中枢节点发现、基于最大连通图的社区发现、基于三角形计数的关系衡量、基于随机游走的用户属性传播等。 **4. 优酷土豆** **“**优酷土豆在使用Hadoop集群的突出问题主要包括：第一是商业智能BI方面，分析师提交任务之后需要等待很久才得到结果;第二就是大数据量计算，比如进行一些模拟广告投放之时，计算量非常大的同时对效率要求也比较高，最后就是机器学习和图计算的迭代运算也是需要耗费大量资源且速度很慢。 最终发现这些应用场景并不适合在MapReduce里面去处理。通过对比，发现Spark性能比MapReduce提升很多。首先，交互查询响应快，性能比Hadoop提高若干倍;模拟广告投放计算效率高、延迟小(同hadoop比延迟至少降低一个数量级);机器学习、图计算等迭代计算，大大减少了网络传输、数据落地等，极大的提高的计算性能。目前Spark已经广泛使用在优酷土豆的视频推荐(图计算)、广告业务等。 ### Spark的战线 Ø  **DAG执行引擎**：不少框架都提出了类似的基于DAG图的执行引擎，来对MapReduce模型的缺点进行优化。如Apache下的Ooize和Tez。 Ø  **内存计算**：作为Spark的杀手锏，分布式内存计算的市场竞争也非常激烈，有非常火的SAP的HANA平台，微软的Dryad，以及Druid。此外，还有众多的数据库厂商推出了内存数据库、内存表或内存计算网格(data grid)等产品。 Ø **交互式查询(Shark)**：Hive的性能问题和(近)实时分析的需求使得不少公司都提出了自己的解决方案。其中最耀眼的当属无敌的Google提出的Dremel嵌套计算模型。而其开源版本也是众多，有Cloudera Impala配合Parquet，以及Apache的Drill。 Ø  **流式计算(Streaming Spark)**：Spark通过将连续的数据流划分成离散的数据段，从而将触角伸到了流式处理领域。而在这一领域，Twitter的Storm(已捐给Apache)被许多公司采用，也是不可小觑。相比之下，Yahoo的S4就显得冷清多了。 Ø **图计算(GraphX)**：图计算方面，Spark借鉴了Pregel和PowerGraph中的图分布式计算和图切割思想，提出了自己的工具包GraphX。它扩展了Spark，例如用EdgeRDD了和VertextRDD表示边和顶点等。 Ø  **机器学习(MLBase)**：机器学习方面，Mahout是个有力的竞争对手，但Mahout也是基于MapReduce任务实现的。而在线机器学习框架Jubatus资料很少，进展不明。 看完这些总结后，不得不佩服Spark的应用范围之广！感觉目前Spark唯一没有单独建立子项目的就是存储方面了，主要是中间内存计算时的存储引擎。目前要么完全由Spark在内存中管理，要么通过Tachyon统一管理。 ### Spark的未来 **Spark 1.0.1** Ø  Spark SQL(原Shark)支持JSON **Spark 1.1** Ø  通用Shuffle接口 Ø  MLLib统计算法 Ø  JDBC驱动 Ø  基于排序的Shuffle **Spark 1.2** Ø  重构存储支持 **Spark 1.3+** Ø  **SparkR对R语言统计分析的支持**(目前已能下载)  ### 参考资料 1 The State of Spark 2[大数据计算新贵Spark在腾讯雅虎优酷成功应用解析](http://www.beagledata.com/news/764.html) 3 The Future of Apache Spark

### 1背景介绍 现今分布式计算框架像MapReduce和Dryad都提供了高层次的原语，使用户不用操心任务分发和错误容忍，非常容易地编写出并行计算程序。然而这些框架都缺乏对分布式内存的抽象和支持，使其在某些应用场景下不够高效和强大。RDD(Resilient Distributed Datasets弹性分布式数据集)模型的产生动机主要来源于两种主流的应用场景： Ø  迭代式算法：迭代式机器学习、图算法，包括PageRank、K-means聚类和逻辑回归(logistic regression) Ø  交互式数据挖掘工具：用户在同一数据子集上运行多个Adhoc查询。 不难看出，这两种场景的共同之处是：**在多个计算或计算的多个阶段间，重用中间结果**。不幸的是，在目前框架如MapReduce中，要想在计算之间重用数据，唯一的办法就是把数据保存到外部存储系统中，例如分布式文件系统。这就导致了巨大的数据复制、磁盘I/O、序列化的开销，甚至会占据整个应用执行时间的一大部分。 为了解决这种问题，研究人员为有这种数据重用需要的应用开发了特殊的框架。例如将中间结果保存在内存中的迭代式图计算框架Pregel。然而这些框架只支持一些特定的计算模式，而没有提供一种通用的数据重用的抽象。于是，RDD横空出世，它的主要功能有： Ø  高效的错误容忍 Ø  中间结果持久化到内存的并行数据结构 Ø  可控制数据分区来优化数据存储 Ø  丰富的操作方法 对于设计RDD来说，最大的挑战在于**如何提供高效的错误容忍(fault-tolerance)**。现有的集群上的内存存储抽象，如分布式共享内存、key-value存储、内存数据库以及Piccolo等，都提供了对可变状态(如数据库表里的Cell)的细粒度更新。在这种设计下为了容错,就必须在集群结点间进行数据复制(data replicate)或者记录日志。这两种方法对于数据密集型的任务来说开销都非常大，因为需要在结点间拷贝大量的数据，而网络带宽远远低于RAM。       与这些框架不同，RDD提供**基于粗粒度转换(coarse-grained transformation)的接口**,例如map、filter、join，能够将同一操作施加到许多数据项上。于是通过记录这些构建数据集(lineage世族)的粗粒度转换的日志，而非实际数据，就能够实现高效的容错。当某个RDD丢失时，RDD有充足的关于丢失的那个RDD是如何从其他RDD产生的信息，从而通过重新计算来还原丢失的数据，避免了数据复制的高开销。       尽管基于粗粒度转换的接口第一眼看起来有些受限、不够强大，但实际上RDD却能很好地用于许多并行计算应用，因为**这些应用本身自然而然地就是在多个数据项上运用相同的操作**。事实上，RDD能够高效地表达许多框架的编程模型，如MapReduce、DryadLINQ、SQL、Pregel和HaLoop，以及它们处理不了的交互式数据挖掘应用。 ### 2 RDD简介 ### 2.1概念 RDD是一种只读的、分区的记录集合。具体来说，RDD具有以下一些特点： Ø  创建：只能通过转换(**transformation**，如map/filter/groupBy/join等，区别于动作action)从两种数据源中创建RDD：1）稳定存储中的数据；2）其他RDD。 Ø  只读：状态不可变，不能修改 Ø  分区：支持使RDD中的元素根据那个key来分区(**partitioning**)，保存到多个结点上。还原时只会重新计算丢失分区的数据，而不会影响整个系统。 Ø  路径：在RDD中叫世族或血统(**lineage**)，即RDD有充足的信息关于它是如何从其他RDD产生而来的。 Ø  持久化：支持将会·被重用的RDD缓存(如in-memory或溢出到磁盘) Ø  延迟计算：像DryadLINQ一样，Spark也会延迟计算RDD，使其能够将转换管道化(pipeline transformation) Ø  操作：丰富的动作(**action**)，count/reduce/collect/save等。 关于转换(transformation)与动作(action)的区别，前者会生成新的RDD，而后者只是将RDD上某项操作的结果返回给程序，而不会生成新的RDD：  ### 2.2例子 假设网站中的一个WebService出现错误，我们想要从数以TB的HDFS日志文件中找到问题的原因，此时我们就可以用Spark加载日志文件到一组结点组成集群的RAM中，并交互式地进行查询。以下是代码示例：        首先行1从HDFS文件中创建出一个RDD，而行2则衍生出一个经过某些条件过滤后的RDD。行3将这个RDD errors缓存到内存中，然而第一个RDD lines不会驻留在内存中。这样做很有必要，因为errors可能非常小，足以全部装进内存，而原始数据则会非常庞大。经过缓存后，现在就可以反复重用errors数据了。我们这里做了两个操作，第一个是统计errors中包含MySQL字样的总行数，第二个则是取出包含HDFS字样的行的第三列时间，并保存成一个集合。        这里要注意的是前面曾经提到过的Spark的延迟处理。Spark调度器会将filter和map这两个转换保存到管道，然后一起发送给结点去计算。 ### 2.3优势 RDD与DSM(distributed shared memory)的最大不同是：RDD只能通过粗粒度转换来创建，而DSM则允许对每个内存位置上数据的读和写。在这种定义下，DSM不仅包括了传统的共享内存系统，也包括了像提供了共享DHT(distributed hash table)的Piccolo以及分布式数据库等。所以RDD相比DSM有着下面这些优势： Ø  高效的容错机制：没有检查点(checkpoint)开销，能够通过世族关系还原。而且还原只涉及了丢失数据分区的重计算，并且重算过程可以在不同结点并行进行，而无需回滚整个系统。 Ø  结点落后问题的缓和(mitigate straggler)：RDD的不可变性使得系统能够运行类似MapReduce备份任务，来缓和慢结点。这在DSM系统中却难以实现，因为多个相同任务一起运行会访问同样的内存数据而相互干扰。 Ø  批量操作：任务能够根据数据本地性(data locality)被分配，从而提高性能。 Ø  优雅降级(degrade gracefully)：当内存不足时，大分区会被溢出到磁盘，提供与其他现今的数据并行计算系统类似的性能。 ### 2.4应用场景 RDD最适合那种在数据集上的所有元素都执行相同操作的批处理式应用。在这种情况下，RDD只需记录世族图谱中的每个转换就能还原丢失的数据分区，而无需记录大量的数据操作日志。所以**RDD不适合那些需要异步、细粒度更新状态的应用**，比如Web应用的存储系统，或增量式的Web爬虫等。对于这些应用，使用具有事务更新日志和数据检查点的数据库系统更为高效。 ### 3 RDD表现形式 ### 3.1深入RDD 使用RDD作为抽象的一个挑战就是：选择一种合适的表现形式，来追踪横跨众多转换的RDD世族关系。在Spark中，我们使用一种简单的、基于图的表现形式，使得Spark在无需为每个转换都增加特殊的处理逻辑的情况下，就能支持大量的转换类型，这大大简化了系统的设计。       总的来说，对于每个RDD都包含五部分信息，即数据分区的集合，能根据本地性快速访问到数据的偏好位置，依赖关系，计算方法，是否是哈希/范围分区的元数据：  以Spark中内建的几个RDD举例来说：

信息/RDD	HadoopRDD	FilteredRDD	JoinedRDD
Partitions	每个HDFS块一个分区，组成集合	与父RDD相同	每个Reduce任务一个分区
PreferredLoc	HDFS块位置	无(或询问父RDD)	无
Dependencies	无(父RDD)	与父RDD一对一	对每个RDD进行混排
Iterator	读取对应的块数据	过滤	联接混排的数据
Partitioner	无	无	HashPartitioner

### 3.2工作原理 在了解了RDD的概念和内部表现形式之后，那么RDD是如何运行的呢？总高层次来看，主要分为三步：创建RDD对象，DAG调度器创建执行计划，Task调度器分配任务并调度Worker开始运行。  以下面一个按A-Z首字母分类，查找相同首字母下不同姓名总个数的例子来看一下RDD是如何运行起来的。  **步骤1：创建RDD。**上面的例子除去最后一个collect是个动作，不会创建RDD之外，前面四个转换都会创建出新的RDD。因此第一步就是创建好所有RDD(内部的五项信息)。 **步骤2：创建执行计划。**Spark会尽可能地管道化，并基于是否要重新组织数据来划分**阶段(stage)**，例如本例中的groupBy()转换就会将整个执行计划划分成两阶段执行。最终会产生一个**DAG(directed acyclic graph，有向无环图)**作为逻辑执行计划。  **步骤3：调度任务。**将各阶段划分成不同的**任务(task)**，每个任务都是数据和计算的合体。在进行下一阶段前，当前阶段的所有任务都要执行完成。因为下一阶段的第一个转换一定是重新组织数据的，所以必须等当前阶段所有结果数据都计算出来了才能继续。       假设本例中的hdfs://names下有四个文件块，那么HadoopRDD中partitions就会有四个分区对应这四个块数据，同时preferedLocations会指明这四个块的最佳位置。现在，就可以创建出四个任务，并调度到合适的集群结点上。  ### 3.3混排 (待补充：关于混排(Shuffle)是如何执行的) ### 3.4宽窄依赖 在设计RDD的接口时，一个有意思的问题是如何表现RDD之间的依赖。在RDD中将依赖划分成了两种类型：窄依赖(narrow dependencies)和宽依赖(wide dependencies)。窄依赖是指**父RDD的每个分区都只被子RDD的一个分区所使用**。相应的，那么宽依赖就是指父RDD的分区被多个子RDD的分区所依赖。例如，map就是一种窄依赖，而join则会导致宽依赖(除非父RDD是hash-partitioned，见下图)。        这种划分有两个用处。首先，窄依赖支持在一个结点上管道化执行。例如基于一对一的关系，可以在filter之后执行map。其次，窄依赖支持更高效的故障还原。因为对于窄依赖，只有丢失的父RDD的分区需要重新计算。而对于宽依赖，一个结点的故障可能导致来自所有父RDD的分区丢失，因此就需要完全重新执行。因此对于宽依赖，Spark会在持有各个父分区的结点上，将中间数据持久化来简化故障还原，就像MapReduce会持久化map的输出一样。 ### 4内部实现 ### 4.1调度器 Spark的调度器类似于Dryad的，但是增加了对持久化RDD分区是否在内存里的考虑。重温一下前面例子里介绍过的：调度器会根据RDD的族谱创建出分阶段的DAG；每个阶段都包含尽可能多的具有窄依赖的变换；具有宽依赖的混排操作是阶段的边界；调度器根据数据本地性分派任务到集群结点上。 ### 4.2解释器集成 (待补充) ### 4.3内存管理 Spark支持三种内存管理方式：Java对象的内存存储，序列化数据的内存存储，磁盘存储。第一种能提供最快的性能，因为JVM能够直接访问每个RDD对象。第二种使用户在内存空间有限时，能选择一种比Java对象图更加高效的存储方式。第三种则对大到无法放进内存，但每次重新计算又很耗时的RDD非常有用。 同时，当有新的RDD分区被计算出来而内存空间又不足时，Spark使用LRU策略将老分区移除到磁盘上。 ### 4.4检查点支持 尽管RDD的Lineage可以用来还原数据，但这通常会非常耗时。所以将某些RDD持久化到磁盘上会非常有用，例如前面提到过的，宽依赖的中间数据。对于Spark来说，对检查点的支持非常简单，因为RDD都是不可变的。所以完全可以在后台持久化RDD，而无需暂停整个系统。 ### 5高级特性 (待补充：Broadcast…) ### 6参考资料 本文内容主要来源于：1）RDD论文《Resilient Distributed Datasets: A Fault-Tolerant Abstraction for In-Memory Cluster Computing》；2）Spark峰会ppt资料：《A-Deeper-Understanding-of-Spark-Internals》和《Introduction to Spark Internals》。感兴趣的可以自行查找。

### ### 1什么是GemFire GemFire是一个位于应用集群和后端数据源之间的高性能、分布式的操作数据(operational data)管理基础架构。它提供了低延迟、高吞吐量的数据共享和事件分发。GemFire充分利用网络中的内存和磁盘资源，形成一个实时的数据网格(data fabric or grid)。  GemFire的主要特性有： Ø  多种网络拓扑 Ø  高并发的内存数据结构，避免锁争夺 Ø  可选的ACID Ø  序列化(native serialization)和智能缓冲(smart buffering)保证消息快速分发 Ø  同步或异步写磁盘 Ø  冗余内存拷贝 ### 2网络拓扑和缓存架构 考虑到问题多样性和架构灵活性，GemFire提供了多种选项来配置在哪(where)以及怎样(how)管理缓存数据，这就使架构师能够从P2P(peer-to-peer)、CS(client-server)、WAN三种组件构建出合适的缓存架构。 ### 2.1 P2P拓扑 在P2P分布式系统中，应用程序使用GemFire的镜像(mirroring)功能来将大量数据跨结点分区(sharding)以及在这些结点间进行数据复制同步。下面主要讲一下GemFire的P2P拓扑中的两个主要角色：**mirrored镜像结点和partitioned分区结点**(具体见3.2中mirror-type的配置方式)。 因为在P2P拓扑中缓存数据与应用在一起，所以首先说一下嵌入式缓存。所谓嵌入式缓存(embedded cache)其实就是说缓存和应用程序在一起，直接利用应用服务器的内存空间。也就是我们常说的类似Ehcache的那种本地缓存(local cache)。  **mirrored结点**就像一块磁铁一样，将其他数据区域的数据都吸附过来，形成一块完整的数据集合。当一块数据区域被配置为mirrored的结点第一次新建或重建时，GemFire将自动执行*初始镜像抓取(initial image fetch)*操作，从其他结点的数据子集中还原出完整的状态。如果此时网络中存在另一个mirrored结点，那么将会执行*最优直接抓取(optimal directed fetch)*。  所以我们很容易看出，mirrored结点主要出于两种目的： Ø  对于大量读的应用，应用程序通过保存全量数据，使客户端请求可以即时访问到想要数据，而无需经过网络传输 Ø  当发生故障时，mirrored结点可以用来恢复其他结点 不同于mirrored结点，每个**partitioned结点**都持有唯一的一块数据。应用程序就像操作本地数据一样，GemFire在幕后管理各个分区的数据，并且保证在至多一跳内(at most one network hop)完成数据访问。根据GemFire的哈希算法，分区数据会被自动放入到各个结点的bucket中。同时GemFire也会自动分配出冗余数据的位置并进行复制。当某个结点出错时，客户端请求会自动被重定向到备份结点。并且GemFire会重新复制出一份数据，从而保证数据的冗余拷贝数。最后，我们可以随时向网络中加入新的结点来对GemFire集群进行动态扩容。  P2P系统提供了低延迟、单跳(one-hop)数据访问、动态发现以及透明化的数据存储位置。但是，网络中的每个结点都要维持一个socket连接到其他每个结点。当结点增多时，连接数将成指数级增长。为了提高扩展性，GemFire提供了一种可靠的UDP多播的通信方式。在下一节中我们将看到，P2P数据同步在服务器间复制数据时的作用。 ### 2.2 Client-Server拓扑 Client-Server缓存允许大量结点相连形成客户端-服务器结构。服务器即为客户端提供缓存，也可以为其他服务器提供数据复制或缓存。  ### 2.3 WAN拓扑 P2P集群由于点和点之间的紧耦合而产生了扩展性问题，这种问题在数据中心有多个集群或数据中心跨城市时被放大。GemFire提供另一种模型来解决。  ### 3 GemFire工作原理 ### 3.1发现机制 默认GemFire使用IP多播来发现新成员，然而所有成员间的通信都采用TCP。对于部署环境禁止使用IP多播或者网络跨越多个子网时，GemFire提供备用方法：使用轻量级的定位服务器(locator server)来追踪所有成员的连接。新成员加入集群时，将询问定位服务并建立类似于IP多播的socket到socket的TCP连接。 ### 3.2数据分发 每个成员都会创建一个或多个缓存数据区域(data region)，通过区域的划分，我们能给每个区域配置不同的分发属性、内存管理以及数据一致性模型。默认GemFire使用P2P分发模型，每个成员都能和其他任何成员通信。同时根据不同的内网特点，传输层可选TCP/IP或可靠多播(UDP)。在这些配置中，有两个属性很重要，**范围(scope)和镜像类型(mirror-type)。** 首先，范围(scope)有四种选项： Ø  Local：不分发。那为什么不直接保存到HashMap中。因为GemFire额外提供了数据自动持久化到磁盘、OQL(Object Query Language)查询数据、数据操作的事务等特性。 Ø  Distribute-no-ack：发送数据给成员1，在发送数据给成员2时不等待成员1的响应。适用于对数据一致性要求不高，并要求低网络延迟的情况。这是GemFire的默认配置，能够提供低延迟、高吞吐，并通过尽快分发来降低数据冲突的概率。 Ø  Distribute-ack：在发送给成员2前，发送数据并等待成员1的响应。这样每条数据都是同步分发的。 Ø  Global：分发前在其他成员上获得锁，再分发数据。适用于悲观的应用场景，通过全局锁服务来管理锁的获得、释放和超时。  现在来看一下第二个重要的配置属性镜像类型(mirror-type)： Ø  none：仅当缓存中有此数据时才更新，任何其他成员发来的新数据都会被忽略掉。适用于某一数据区域仅用来保存另一区域数据的子集。 Ø  keys：数据区域仅保存key来节约内存，当真正有请求时再从其他区域抓取数据并保存到本地，之后接受对此数据项的更新。适用于无法预测哪些数据会被某一结点访问的情况。 Ø  keys-values：真正的镜像，将保存全量数据。适用于需要立即访问所有数据的结点，以及数据冗余备份。 这两个属性的配置对数据区域中保存的是什么数据有很大影响：  ### 4持久化和溢出 持久化(persistence)将整个数据集拷贝到磁盘，当成员出错时可以用来还原数据。而溢出(overflow)保存key在内存中而value保存到磁盘，达到节省内存的目的。两者既可以单独使用，也可以混合使用。 ### 4.1持久化 GemFire支持两种写磁盘选项：操作内存数据时同步写，或者固定间隔异步写。后一种只当应用在出错时能够容忍不完整的数据还原时使用。  ### 4.2溢出 当内存不足时，GemFire使用LRU策略来决定是否对某个数据项溢出。  ### 4.3混合使用 持久化与溢出可以混合使用。所有key-value都备份到磁盘，并且当内存不足时，只保留最近使用过的数据。由于LRU而被移除到磁盘的value不会对磁盘有影响，因为所有数据已被持久化到磁盘上了。  ### 5事务 GemFire支持缓存事务与JTA事务两种。 ### 5.1缓存事务 每个事务都有其私有的工作区域。事务开始时，数据将被拷贝到私有区域，直到事务提交。若提交时没有冲突，则数据从私有区域拷贝回原区域。这样事务就可以并发地修改缓存了。  对于范围(scope)配置为local的缓存数据区域，事务提交后就算是完成了。但对于分布式(scope=distributed-no-ack or distributed-ack)，则在事务提交时要进行缓存同步。 ### 6查询 (待补充：OOL) ### 7数据可用性和Failover (待补充)

> 原文出处：[内存计算](http://blog.csdn.net/column/details/in-memory-computing.html) 作者：[戴晨](http://blog.csdn.net/dc_726) **本系列文章经作者授权在看云整理发布，未经作者允许，请勿转载！** # 内存计算 > 关注大数据内存计算领域，涉及内存计算框架、内存数据库、内存数据/计算网格等方面，解决海量数据的实时计算、分析问题。