8620-84511745

博客

最新的博客和消息

微服务架构多“微”才合适?

Posted in 其他 on Aug 09, 2016

互联网架构为什么要做服务化?

Docker到底是网红还是趋势?

Posted in docker on Aug 04, 2016

Docker作为近期备受关注的技术,在国内已逐步从观望、试验走向生产交付。

一篇文章,掌握所有开源数据库的现状

Posted in 其他 on Jul 31, 2016

(转载)

数据库作为业务的核心,在整个基础软件栈中是非常重要的一环。近几年社区也是新的方案和思想层出不穷,接下来我将总结一下近几年一些主流的开源数据库方案,其背后的设计思想以及适用场景。本人才疏学浅如有遗漏或者错误请见谅。本次分享聚焦于数据库既结构化数据存储 OLTP 及 NoSQL 领域,不会涉及 OLAP、对象存储、分布式文件系统。

1.开源RDBMS与互联网的崛起

很长时间以来,关系型数据库一直是大公司的专利,市场被 Oracle / DB2 等企业数据库牢牢把持。但是随着互联网的崛起、开源社区的发展,上世纪九十年代 MySQL 1.0 的发布,标志着关系型数据库的领域社区终于有可选择的方案。

MySQL

第一个介绍的单机RDBMS就是 。相信大多数朋友都已经对 MySQL 非常熟悉,基本上 MySQL 的成长史就是互联网的成长史。我接触的第一个 MySQL 版本是 MySQL 4.0,到后来的 MySQL 5.5 更是经典——基本所有的互联网公司都在使用。

MySQL 也普及了「可插拔」引擎这一概念,针对不同的业务场景选用不同的存储引擎是 MySQL tuning 的一个重要的方式。比如对于有事务需求的场景使用 InnoDB;对于并发读取的场景 MyISAM 可能比较合适;但是现在我推荐绝大多数情况还是使用 InnoDB,毕竟 5.6 后已经成为了官方的默认引擎。大多数朋友都基本知道什么场景适用 MySQL(几乎所有需要持久化结构化数据的场景),我就不赘述了。

另外值得一提的是 MySQL 5.6中引入了多线程复制和 GTID,使得故障恢复和主从的运维变得比较方便。另外,5.7(目前处于 GA 版本) 是 MySQL 的一个重大更新,主要是读写性能和复制性能上有了长足的进步(在5.6版本中实现了SCHEMA级别的并行复制,不过意义不大,倒是MariaDB的多线程并行复制大放异彩,有不少人因为这个特性选择MariaDB。MySQL 5.7 MTS支持两种模式,一种是和5.6一样,另一种则是基于binlog group commit实现的多线程复制,也就是MASTER上同时提交的binlog在SLAVE端也可以同时被apply,实现并行复制)。

如果有 单机数据库技术选型 的朋友,基本上只需要考虑 5.7 或者 MariaDB 就好了,而且 5.6、5.7 由 Oracle 接手后,性能和稳定性上都有了明显的提升。

PostgreSQL

的历史也非常悠久,其前身是 的 ,主持这个项目的 Michael Stronebraker 于 2015 年获得图灵奖。后来项目更名为 Post-Ingres,项目基于 BSD license 下开源。 1995 年几个 UCB 的学生为 Post-Ingres 开发了 SQL 的接口,正式发布了 PostgreSQL95,随后一步步在开源社区中成长起来。

和 MySQL 一样,PostgreSQL 也是一个单机的关系型数据库,但是与 MySQL 方便用户过度扩展的 SQL 文法不一样的是,PostgreSQL 的 SQL 支持非常强大,不管是内置类型、JSON 支持、GIS 类型以及对于复杂查询的支持,PL/SQL 等都比 MySQL 强大得多。而且从代码质量上来看,PostgreSQL 的代码质量是优于 MySQL 的,另外 PostgreSQL 的 SQL 优化器比 MySQL 强大很多,几乎所有稍微复杂的查询(当然,我没有对比 MySQL 5.7,也可能这个信息 outdated 了)PostgreSQL 的表现都优于 MySQL。

从近几年的趋势上来看, PostgreSQL 的势头也很强劲 ,我认为 PostgreSQL 的不足之处在于没有 MySQL 这样强大的社区和群众基础。MySQL 经过那么多年的发展,积累了很多的运维工具和最佳实践,但是 PostgreSQL 作为后起之秀,拥有更优秀的设计和更丰富的功能。PostgreSQL 9 以后的版本也足够稳定,在做新项目技术选型的时候,是一个很好的选择。另外也有很多新的数据库项目是基于 PostgreSQL 源码的基础上进行二次开发,比如 等。

我认为,单机数据库的时代很快就会过去 。榨取摩尔定律带来的硬件红利总是有上限的,现代业务的数据规模、流量以及现代的数据科学对于数据库的要求单机已经很难满足。网卡磁盘 IO 和 CPU 总有瓶颈,线上敏感的业务系统可能还得承担 SPOF(单点故障) 的风险,主从复制模型在主挂掉时到底切还是不切?切了以后数据如何恢复?如果只是出现主从机器网络分区问题呢?甚至是监控环境出现网络分区问题呢?这些都是问题

所以我的观点是,无论单机性能多棒(很多令人乍舌的评测数据都是针对特定场景的优化,另外甚至有些都是本机不走网络,而大多数情况数据库出现的第一个瓶颈其实是网卡和并发连接……),随着互联网的蓬勃发展,移动互联网的出现使得数据库系统迎来了第一次分布式的洗礼。

2. 分布式时代:NoSQL的复兴和模型简化的力量

在介绍 NoSQL 之前,我想提两个公司,一个是Google,另一个是Amazon。

Google

Google 应该是第一个将分布式存储技术应用到大规模生产环境的公司,同时也是在分布式系统上积累最深的公司,可以说目前工业界的分布式系统的工程实践及思想大都来源于 Google。比如 2003 年的 GFS 开创了分布式文件系统,2006 年的 Bigtable 论文开创了分布式键值系统,直接催生的就是 Hadoop 的生态;至于 2012 年发表论文的 和 更是一个指明未来关系型数据库发展方向的里程碑式的项目,这个我们后续会说。

Amazon

另一个公司是 Amazon。2007 年发表的 尝试引入了最终一致性的概念, WRN 的模型及向量时钟的应用,同时将一致性 HASH、merkle tree 等当时一些很新潮的技术整合起来,正式标志着 NoSQL 的诞生——对后来业界的影响也是很大,包括后来的 Cassandra、RiakDB、Voldemort 等数据库都是基于 Dynamo 的设计发展起来的。

新思潮

另外这个时期(2006 年前后持续至今)一个比较重要的思潮就是 数据库(持久化)和缓存开始有明确的分离 ——我觉得这个趋势是从 memcached 开始的。随着业务的并发越来越高,对于低延迟的要求也越来越高;另外一个原因是随着内存越来越便宜,基于内存的存储方案渐渐开始普及。当然内存缓存方案也经历了一个从单机到分布式的过程,但是这个过程相比关系型数据库的进化要快得多。

这是因为 NoSQL 的另外一个重要的标志—— 数据模型的变化 ——大多 NoSQL 都抛弃了关系模型,选择更简单的键值或者文档类型进行存储。数据结构和查询接口都相对简单,没有了SQL 的包袱,实现的难度会降低很多。

另外 NoSQL 的设计几乎都选择牺牲掉复杂 SQL 的支持及 ACID 事务换取弹性扩展能力,也是从当时互联网的实际情况出发:业务模型简单、爆发性增长带来的海量并发及数据总量爆炸、历史包袱小、工程师强悍,等。其中最重要的还是业务模型相对简单。

嵌入式存储引擎

在开始介绍具体的开源的完整方案前,我想介绍一下嵌入式存储引擎们。

随着 NoSQL 的发展,不仅仅缓存和持久化存储开始细分,再往后的存储引擎也开始分化并走上前台。之前很难想象一个存储引擎独立于数据库直接对外提供服务,就像你不会直接拿着 InnoDB 或者 MyISAM甚至一个 B-tree 出来用一样(当然,bdb 这样鼎鼎大名的除外)。人们基于这些开源的存储引擎进行进一步的封装,比如加上网络协议层、加上复制机制等等,一步步构建出完整的风格各异的 NoSQL 产品。

这里我挑选几个比较 著名存储引擎 介绍一下。

TC

我最早接触的是 。TC 相信很多人也都听说过,TC 是由日本最大的社交网站 Mixi 开发并开源的一个混合 Key-Value 存储引擎,其中包括 HASH Table 和 B+ Tree 的实现。但是这个引擎的一个缺陷是随着数据量的膨胀,性能的下降会非常明显,而且现在也基本不怎么维护了,所以入坑请慎重。于 TC 配合使用的 是一个网络库,为 TC 提供网络的接口使其变成一个数据库服务,TT + TC 应该是比较早的 NoSQL 的一个尝试。

LevelDB

在 2011 年,Google 开源了 Bigtable 的底层存储擎: 。LevelDB 是一个使用 C++ 开发的嵌入式的 Key-Value 存储引擎,数据结构采用了 LSM-Tree,具体 LSM-Tree 的算法分析可以很容易在网上搜索到,我就不赘述了。其特点是,对于写入极其友好,LSM 的设计避免了大量的随机写入;对于特定的读也能达到不错的性能(热数据在内存中);另外 LSM-Tree 和 B-tree 一样是支持有序 Scan 的;而且 LevelDB 是出自 Jeff Dean 之手,他的事迹做分布式系统的朋友一定都知道,不知道的可以去 Google 搜一下。

LevelDB 拥有极好的写性能,线程安全,BaTCh Write 和 Snapshot 等特性,使其很容易的在上层构建 MVCC 系统或者事务模型,对于数据库来说非常重要。

另外值得一说的是,Facebook 维护了一个活跃的 LevelDB 的分支,名为 RocksDB。RocksDB 在 LevelDB 上做了很多的改进,比如多线程 Compactor、分层自定义压缩、多 MemTable 等。另外 RocksDB 对外暴露了很多 Configration ,可以根据不同业务的形态进行调优;同时 Facebook 在内部正在用 RocksDB 来实现一个全新的 MySQL 存储引擎:MyRocks,值得关注。RocksDB 的社区响应速度很快也很友好,实际上 PingCAP 也是 RocksDB 的社区贡献者。我建议新的项目如果在 LevelDB 和 RocksDB 之间纠结的话,请果断选择 RocksDB。

B-tree 家族

当然,除了 LSM-Tree 外, 的家族也还是有很多不错的引擎。首先大多数传统的单机数据库的存储引擎都选择了 ,B+Tree 对磁盘的读比较友好,第三方存储引擎比较著名的纯 B+Tree 实现是 。首先 LMDB 选择在内存映像文件 (mmap) 实现 B+Tree,同时使用了 Copy-On-Write 实现了 MVCC 实现并发事务无锁读的能力,对于高并发读的场景比较友好;同时因为使用的是 mmap 所以拥有跨进程读取的能力。因为我并没有在生产环境中使用过 LMDB ,所以并不能给出 LMDB 的一些缺陷,见谅。

混合引擎

还有一部分的存储引擎选择了多种引擎混合,比如最著名的应该是 ,大概是去年被 MongoDB 收购,现在成为了 MongoDB 的默认存储引擎。WiredTiger 内部有 LSM-Tree 和 B-tree 两种实现提供一套接口,根据业务的情况可自由选择。另外一些特殊数据结构的存储引擎在某些特殊场合下非常抢眼,比如极高压缩比 ,采用了名为分形树的数据结构,在维持一个可接受的读写压力的情况下,能拥有 10 倍以上的压缩率。

NoSQL

说完了几个比较著名的存储引擎,我们来讲讲比较著名的 NoSQL。在我的定义中,NoSQL 是Not Only SQL 的缩写,所以可能包含的范围有内存数据库,持久化数据库等。总之就是和单机的关系型数据库不一样的结构化数据存储系统。

我们先从缓存开始。

memcached

前面提到了 memcached 应该是第一个大规模在业界使用的缓存数据库,memcached 的实现极其简单,相当于将内存用作大的 HASH Table,只能在上面 get/set/ 计数器等操作,在此之上用 libevent 封装了一层网络层和文本协议(也有简单的二进制协议),虽然支持一些 CAS 的操作,但是总体上来看,还是非常简单的。

但是 memcached 的 内存利用率并不太高 ,这个因为 memcached 为了避免频繁申请内存导致的内存碎片的问题,采用了自己实现的slab allocator 的方式。即内存的分配都是一块一块的,最终存储在固定长度的chunk 上,内存最小的分配单元是chunk,另外 libevent 的性能也并没有优化到极致,但是不妨碍 memcached 成为当时的开源缓存事实标准(另外,八卦一下,memcached 的作者 现在在 Google,大家如果用 Golang 的话,Go 的官方 HTTP 包就是这哥们写的,是个很高产的工程师)。

Redis

如果我没记错的话,在 2009 年前后,一位意大利的工程师 ,开源了 。从此彻底颠覆了缓存的市场, 到现在大多数缓存的业务都已用上Redis,memcached 基本退出了历史舞台 。Redis 最大的特点是拥有丰富的数据结构支持,不仅仅是简单的 Key-Value,包括队列、集合、Sorted Set 等等,提供了非常丰富的表达力,而且 Redis 还提供 sub/pub 等超出数据库范畴的便捷功能,使得几乎一夜之间大家纷纷投入 Redis 的怀抱。

Twemproxy

但是随着 Redis 渐渐的普及,而且越用越狠,另外内存也越来越便宜,人们开始寻求 扩展单机Redis的方案 ,最早的尝试是twitter 开源的 ,twemproxy 是一个 Redis 中间件,基本只有最简单的数据路由功能,并没有动态的伸缩能力,但是还是受到了很多公司的追捧,因为确实没方案。 随后的 Redis Cluster 也是难产了好久,时隔好几年,中间出了 7 个RC 版本,最后才发布;

2014 年底,我们开源了 ,解决了 Redis 中间件的数据弹性伸缩问题,目前广泛应用于国内各大互联网公司中,这个在网上也有很多文章介绍,我也就不展开了。 所以在缓存上面,开源社区现在倒是非常统一,就是 Redis 极其周边的扩展方案 。

MongoDB

在 NoSQL 的大家庭中, 其实是一个异类,大多 NoSQL 舍弃掉 SQL 是为了追求更极致的性能和可扩展能力,而 MongoDB 主动选择了文档作为对外的接口,非常像 JSON 的格式。Schema-less 的特性对于很多轻量级业务和快速变更了互联网业务意义很大,而且 MongoDB 的易用性很好,基本做到了开箱即用,开发者不需要费心研究数据的表结构,只需要往里存就好了,这确实笼络了一大批开发者。

尽管 MongoDB 早期的版本各种不稳定,性能也不太好(早期的 Mongo 并没有存储引擎,直接使用了 mmap 文件),集群模式还全是问题(比如至今还未解决的 Cluster 同步带宽占用过多的问题),但是因为确实太方便了,在早期的项目快速迭代中,Mongo 是一个不错的选择。

但是这也正是它的问题,我不止一次听到当项目变得庞大或者「严肃」的时候, 团队最后还是回归了关系型数据库 。Anyway,在 2014 年底 MongoDB 收购了 WiredTiger 后,在 2.8 版本中正式亮相,同时 3.0 版本后更是作为默认存储引擎提供,性能和稳定性有了非常大的提升。

但是,从另一方面讲,Schema-less 到底对软件工程是好事还是坏事这个问题还是有待商榷。我个人是站在 Schema 这边的,不过在一些小项目或者需要快速开发的项目中使用 Mongo 确实能提升很多的开发效率,这是毋庸置疑的。

HBase

说到 NoSQL 不得不提的是 ,HBase 作为Hadoop 旗下的重要产品, 的正统开源实现,是不是有一种钦定的感觉:)。提到 HBase 就不得不提一下 ,Bigtable是Google内部广泛使用的分布式数据库,接口也不是简单的Key-Value,按照论文的说法叫:multi-dimensional sorted map,也就是 Value 是按照列划分的。Bigtable 构建在 GFS 之上,弥补了分布式文件系统对于海量、小的、结构化数据的插入、更新、随机读请求的缺陷。

HBase 就是这么一个系统的实现, 底层依赖 HDFS 。HBase 本身并不实际存储数据,持久化的日志和 SST file (HBase 也是 LSM-Tree 的结构) 直接存储在 HDFS 上,Region Server (RS) 维护了 MemTable 以提供快速的查询,写入都是写日志,后台进行 Compact,避免了直接随机读写 HDFS。

数据通过 Region 在逻辑上进行分割,负载均衡通过调节各个 Region Server 负责的 Region 区间实现。当某 Region 太大时,这个 Region 会分裂,后续可能由不同的 RS 负责,但是前面提到了,HBase 本身并不存储数据,这里的 Region 仅是逻辑上的,数据还是以文件的形式存储在 HDFS 上,所以 HBase 并不关心 Replication 、水平扩展和数据的分布,统统交给 HDFS 解决。

和 Bigtable 一样,HBase 提供行级的一致性,严格来说在 中它是一个 CP 的系统,但遗憾的是并没有更进一步提供 ACID 的跨行事务。HBase 的好处就不用说了,显而易见,通过扩展 RS 可以几乎线性提升系统的吞吐,及 HDFS 本身就具有的水平扩展能力。

但是缺点仍然是有的 。

首先,Hadoop 的软件栈是 Java,JVM 的 GC Tuning 是一个非常烦人的事情,即使已经调得很好了,平均延迟也得几十毫秒;

另外在架构设计上,HBase 本身并不存储数据,所以可能造成客户端请求的 RS 并不知道数据到底存在哪台 HDFS DataNode 上,凭空多了一次 RPC;

第三,HBase 和 Bigtable 一样,并不支持跨行事务,在 Google 内部不停的有团队基于 Bigtable 来做分布式事务的支持,比如 MegaStore、Percolator。后来 有次接受 也提到非常后悔没有在 Bigtable 中加入跨行事务,不过还好这个遗憾在 Spanner 中得到了弥补,这个一会儿说。

总体来说,HBase 还是一个非常健壮且久经考验的系统,但是需要你有对于 Java 和 Hadoop 比较深入的了解后,才能玩转,这也是 Hadoop 生态的一个问题,易用性真是不是太好,而且社区演进速度相对缓慢,也是因为历史包袱过重的缘故吧。

Cassandra

提到 ( C ),虽然也是 Dynamo 的开源实现,但就没有这种钦定的感觉了。 C 确实命途多舛,最早 2008 由 Facebook 开发并开源,早期的 C* 几乎全是 bug,Facebook 后来索性也不再维护转过头搞 HBase 去了,一个烂摊子直接丢给社区。还好 把这个项目捡起来商业化,搞了两年,终于渐渐开始流行起来。

C 不能简单的归纳为读快写慢,或者读慢写快,因为采用了 qourm 的模型,调整复制的副本数以及读的数量,可以达到不同的效果,对于一致性不是特别高的场景,可以选择只从一个节点读取数据,达到最高的读性能。另外 C 并不依赖分布式文件系统,数据直接存储在磁盘上,各个存储节点之间自己维护复制关系,减少了一层 RPC 调用,延迟上对比 HBase 还是有一定优势的。

不过即使使用 qourm 的模型也并不代表 C 是一个强一致的系统。 C 并不帮你解决冲突,即使你 W(写的副本数) + R(读请求的副本数) > N(节点总数), C 也没办法帮你决定哪些副本拥有更新的版本,因为每个数据的版本是一个 NTP 的时间戳或者客户端自行提供,每台机器可能都有误差,所以有可能并不准确,这也就是为什么 C 是一个 AP 的系统。不过 C* 一个比较友好的地方是提供了 CQL,一个简单的 SQL 方言,比起 HBase 在易用性上有明显优势。

即使作为一个 AP 系统, C 已经挺快了,但是人们追求更高性能的脚步还是不会停止。应该是今年年初, 的发布就是典型的证明,ScyllaDB 是一个兼容 C 的 NoSQL 数据库,不一样的是,ScyllaDB 完全用 C++ 开发,同时使用了类似 DPDK 这样的黑科技,具体我就不展开了,有兴趣可以到 Scylla 的官网去看看。BTW,国内的蘑菇街第一时间使用了 ScyllaDB,同时在 Scylla 的官网上 share 了他们的方案,性能还是很不错的。

3. 中间件与分库分表

NoSQL 就先介绍到这里,接下来我想说的是一些在基于单机关系型数据库之上的中间件和分库分表方案。

在这方面确实历史悠久,而且也是没有办法的选择,关系型数据库不比Redis ,并不是简单的写一个类似Twemproxy 的中间件就搞定了。数据库的中间件需要考虑很多,比如解析 SQL,解析出 sharding key,然后根据 sharding key 分发请求,再合并;另外数据库有事务,在中间件这层还需要维护 Session 及事务状态,而且大多数方案并没有办法支持跨 shard 的事务。

这就不可避免的导致了业务使用起来会比较麻烦,需要重写代码,而且会增加逻辑的复杂度,更别提动态的扩容缩容和自动的故障恢复了。在集群规模越来越大的情况下,运维和 DDL 的复杂度是指数级上升 的。

中间件项目盘点

数据库中间件最早的项目大概是 , 用于实现读写分离 。后来国人在这个领域有过很多的 著名的开源项目,比如阿里的Cobar和DDL(并未完全开源;后来社区基于 Cobar 改进的MyCAT、360 开源的Atlas 等 ,都属于这一类中间件产品;

在中间件这个方案上基本走到头的开源项目应该是 。Vitess 基本上是一个集大成的中间件产品,内置了热数据缓存、水平动态分片、读写分离等等,但是代价也是整个项目非常复杂,另外文档也不太好。大概1年多以前,我们尝试搭建起完整的 Vitess 集群,但是并未成功,可见其复杂度。

另外一个 值得一提 的是 这个项目,Postgres-XC 的野心还是很大的,整体的架构有点像早期版本的 OceanBase,由一个中央节点来处理协调分布式事务 / 解决冲突,数据分散在各个存储节点上,应该是目前 PostgreSQL 社区最好的分布式扩展方案。其他的就不提了。

  1. 未来在哪里?NewSQL?

一句话,NewSQL 是未来。

2012 年 Google 在 OSDI 上发表了 Spanner 的论文,2013 年在 SIGMOD 发表了 F1 的论文。这两篇论文让业界第一次看到了关系模型和 NoSQL 的扩展性在超庞大集群规模上融合的可能性。在此之前,大家普遍认为这个是不可能的,即使是 Google 也经历了 这样系统的失败。

Spanner综述

但是 Spanner 的创新之处在于通过硬件(GPS时钟+原子钟)来解决时钟同步的问题。在分布式系统里,时钟是最让人头痛的问题,刚才提到了 C* 为什么不是一个强 C 的系统,正是因为时钟的问题。而 Spanner 的厉害之处在于即使两个数据中心隔得非常远,不需要有通信(因为通信的代价太大,最快也就是光速)就能保证 TrueTime API的时钟误差在一个很小的范围内(10ms)。另外 Spanner 沿用了很多 Bigtable 的设计,比如 Tablet / Directory 等,同时在 Replica 这层使用 Paxos 复制,并未完全依赖底层的分布式文件系统。但是 Spanner 的设计底层仍然沿用了 Colossus,不过论文里也说是可以未来改进的点。

Google 的内部的数据库存储业务,大多是 3~5 副本,重要一点的 7 副本,遍布全球各大洲的数据中心,由于普遍使用了 Paxos,延迟是可以缩短到一个可以接受的范围(Google 的风格一向是追求吞吐的水平扩展而不是低延迟,从悲观锁的选择也能看得出来,因为跨数据中心复制是必选的,延迟不可能低,对于低延迟的场景,业务层自己解决或者依赖缓存)。

另外由 Paxos 带来的 Auto-Failover 能力,更是能让整个集群即使数据中心瘫痪,业务层都是透明无感知的。另外 F1 构建在 Spanner 之上,对外提供了更丰富的 SQL 语法支持,F1 更像一个分布式 MPP SQL——F1 本身并不存储数据,而是将客户端的 SQL 翻译成类似 MapReduce 的任务,调用 Spanner 来完成请求。

其实除了 TrueTime 整个系统并没有用什么全新的算法,而是近些年分布式系统的技术 Spanner 和 F1 的出现标志着第一个 NewSQL 在生产环境中提供服务。

有以下几个重点:

  • 完整的 SQL 支持,ACID 事务;

  • 弹性伸缩能力;

  • 自动的故障转移和故障恢复,多机房异地灾备。

NewSQL 特性确实非常诱人,在 Google 内部,大量的业务已经从原来的 Bigtable 切换到 Spanner 之上。我相信未来几年,整个业界的趋势也是如此,就像当年的 Hadoop 一样,Google 的基础软件的技术趋势是走在社区前面的。

社区反应

Spanner 的论文发表之后,当然也有社区的追随者开始实现(比如我们 :D ),第一个团队是在纽约的 。CockroachDB 的团队的组成还是非常豪华的,早期团队由是 Google 的分布式文件系统 团队的成员组成;技术上来说,Cockroach 的设计和 Spanner 很像,不一样的地方是没有选择 TrueTime而是 HLC (Hybrid logical clock),也就是 NTP +逻辑时钟来代替 TrueTime 时间戳;另外 Cockroach 选用了 Raft 代替 Paxos 实现复制和自动容灾,底层存储依赖 RocksDB 实现,整个项目使用 Go 语言开发,对外接口选用 PostgreSQL 的 SQL 子集。

CockroachDB

CockroachDB 的技术选型比较激进,比如依赖了 HLC 来做事务的时间戳。但是在 Spanner 的事务模型的 Commit Wait 阶段等待时间的选择,CockroachDB 并没有办法做到 10ms 内的延迟;CockroachDB 的 Commit Wait 需要用户自己指定,但是谁能拍胸脯说 NTP 的时钟误差在多少毫秒内?我个人认为在处理跨洲际机房时钟同步的问题上,基本只有硬件时钟一种办法。HLC 是没办法解决的。

另外Cockroach 采用了 gossip 来同步节点信息,当集群变得比较大的时候,gossip 心跳会是一个非常大的开销。当然 CockroachDB 的这些技术选择带来的优势就是非常好的易用性,所有逻辑都在一个 binary 中,开箱即用,这个是非常大的优点。

TiDB

目前从全球范围来看,另一个在朝着 Spanner / F1 的开源实现这个目标上走的产品是 TiDB(终于谈到我们的产品了)。TiDB 本质上是一个更加正统的 Spanner 和 F1 实现,并不像 CockroachDB 那样选择将 SQL 和 Key-Value 融合,而是像 Spanner 和 F1 一样选择分离,这样分层的思想也是贯穿整个 TiDB 项目始终的。对于测试、滚动升级以及各层的复杂度控制会比较有优势;另外 TiDB 选择了 MySQL 协议和语法的兼容,MySQL 社区的 ORM 框架,运维工具,直接可以应用在 TiDB 上。

和 Spanner一样,TiDB 是一个无状态的 MPP SQL Layer,整个系统的底层是依赖 TiKey-Value 来提供分布式存储和分布式事务的支持。TiKey-Value 的分布式事务模型采用的是 Google Percolator 的模型,但是在此之上做了很多优化。Percolator 的优点是去中心化程度非常高,整个集群不需要一个独立的事务管理模块,事务提交状态这些信息其实是均匀分散在系统的各个 Key 的 meta 中,整个模型唯一依赖的是一个授时服务器。

在我们的系统上,极限情况这个授时服务器每秒能分配 400w 以上个单调递增的时间戳,大多数情况基本够用了(毕竟有 Google 量级的场景并不多见);同时在 TiKey-Value 中,这个授时服务本身是高可用的,也不存在单点故障的问题。

TiKey-Value 和 CockroachDB 一样也是选择了 Raft 作为整个数据库的基础;不一样的是,TiKey-Value 整体采用 Rust 语言开发,作为一个没有 GC 和 Runtime 的语言,在性能上可以挖掘的潜力会更大。

关于未来

我觉得 未来的数据库会有几个趋势 ,也是 TiDB 项目追求的目标:

  • 数据库会随着业务云化,未来一切的业务都会跑在云端,不管是私有云或者公有云,运维团队接触的可能再也不是真实的物理机,而是一个个隔离的容器或者「计算资源」。这对数据库也是一个挑战,因为数据库天生就是有状态的,数据总是要存储在物理的磁盘上,而数据的移动的代价比移动容器的代价可能大很多。

  • 多租户技术会成为标配,一个大数据库承载一切的业务,数据在底层打通,上层通过权限,容器等技术进行隔离;但是数据的打通和扩展会变得异常简单,结合第一点提到的云化,业务层可以再也不用关心物理机的容量和拓扑,只需要认为底层是一个无穷大的数据库平台即可,不用再担心单机容量和负载均衡等问题。

  • OLAP 和 OLTP 会进一步细分,底层存储也许会共享一套,但是SQL优化器这层的实现一定是千差万别的。对于用户而言,如果能使用同一套标准的语法和规则来进行数据的读写和分析,会有更好的体验。

  • 在未来分布式数据库系统上,主从日志同步这样落后的备份方式会被 Multi-Paxos / Raft 这样更强的分布式一致性算法替代,人工的数据库运维在管理大规模数据库集群时是不可能的,所有的故障恢复和高可用都会是高度自动化的。

Spark Streaming kafka 实现数据零丢失的几种方式

Posted in 大数据 on Jul 27, 2016

定义

问题开始之前先解释下流处理中的一些概念:

  • At most once - 每条数据最多被处理一次(0次或1次)
  • At least once - 每条数据最少被处理一次 (1次或更多)
  • Exactly once - 每条数据只会被处理一次(没有数据会丢失,并且没有数据会被多次处理)
High Level API

如果不做容错,将会带来数据丢失

因为receiver一直在接收数据,在其没有处理的时候(已通知zk数据接收到),executor突然挂掉(或是driver挂掉通知executor关闭),缓存在其中的数据就会丢失。

因为这个问题,Spark1.2开始加入了WAL(Write ahead log

开启 WAL,将receiver获取数据的存储级别修改为StorageLevel.MEMORY_AND_DISK_SER

val conf = new SparkConf()

conf.set("spark.streaming.receiver.writeAheadLog.enable","true")

val sc= new SparkContext(conf)

val ssc = new StreamingContext(sc,Seconds(5))

ssc.checkpoint("walDir")

val lines = KafkaUtils.createStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParams, topicMap, StorageLevel.MEMORY_AND_DISKSER).map(._2)

开启WAL后,依旧存在数据丢失问题

即使按官方说的设置了WAL,依旧会有数据丢失,这是为什么?因为在任务中断时receiver也被强行终止了,将会造成数据丢失,提示如下:

ERROR ReceiverTracker: Deregistered receiver for stream 0: Stopped by driver

WARN BlockGenerator: Cannot stop BlockGenerator as its not in the Active state [state = StoppedAll]

WARN BatchedWriteAheadLog: BatchedWriteAheadLog Writer queue interrupted.

在Streaming程序的最后添加代码,只有在确认所有receiver都关闭的情况下才终止程序。

sys.addShutdownHook({ ssc.stop(true,true)})

调用的方法为:

def stop(stopSparkContext: Boolean, stopGracefully: Boolean): Unit

WAL带来的问题

WAL实现的是At-least-once语义。 如果在写入到外部存储的数据还没有将offset更新到zookeeper就挂掉,这些数据将会被反复消费。同时,降低了程序的吞吐量。

Kafka Direct API

Kafka direct API 的运行方式,将不再使用receiver来读取数据,也不用使用WAL机制。

同时保证了exactly-once语义,不会在WAL中消费重复数据。不过需要自己完成将offset写入zk的过程,在官方文档中都有相应介绍。

例如如下的调用方式:

messages.foreachRDD(rdd=>{ val message = rdd.map(_._2) //对数据进行一些操作

message.map(method)//更新zk上的offset (自己实现)

updateZKOffsets(rdd) })

原文链接:http://www.jianshu.com/p/716af5449175

五个技巧教你用编程实现数据可视化

Posted in 其他 on Jul 22, 2016

用编程实现可视化其实是非常有趣的,虽然从起点学习编程不是那么容易,而且大部分人都会说没有足够的时间,但我依然觉得,为了获得长期的收益,从一开始花点功夫还是值得的。

Gartner:使用容器技术比传统架构更安全

Posted in 其他 on Jul 15, 2016

Gartner的研究结果显示,容器比在传统操作系统里运行的应用程序更安全,因此那些不想被黑客攻击的架构应该认真考虑往云里迁移了。

大数据公司 Splunk 和 Cloudera 的核心竞争力在哪里?

Posted in 其他 on Jul 11, 2016

像硅谷这种初创公司Splunk和Cloudera,他们的核心竞争力究竟在哪里?

关于OpenStack的九个关键问题

Posted in Openstack on Jul 07, 2016

,OpenStack 将能成为未来 IT 架构的主流。目前已有许多企业正在寻找属于自己的基础架构技术,希望自家资料中心可以运作得更有效率,让企业更快地前进。而 OpenStack 也因此成为他们的选择,不只是一个开源专案,而逐渐变成了企业的核心。

Redis 和 Memcached 的区别

Posted in Redis on Jul 03, 2016

Redis支持服务器端的数据操作:Redis相比Memcached来说,拥有更多的数据结构和并支持更丰富的数据操作,通常在Memcached里,你需要将数据拿到客户端来进行类似的修改再set回去。这大大增加了网络IO的次数和数据体积。

脱离JVM? Hadoop生态圈的挣扎与演化

Posted in Hadoop on Jun 30, 2016

(转发)

作者:李呈祥

链接:https://zhuanlan.zhihu.com/p/20228397

来源:知乎 / 著作权归作者所有。

新世纪以来,互联网及个人终端的普及,传统行业的信息化及物联网的发展等产业变化产生了大量的数据,远远超出了单台机器能够处理的范围,分布式存储与处理成为唯一的选项。从2005年开始,Hadoop从最初Nutch项目的一部分,逐步发展成为目前最流行的大数据处理平台。Hadoop生态圈的各个项目,围绕着大数据的存储,计算,分析,展示,安全等各个方面,构建了一个完整的大数据生态系统,并有Cloudera,HortonWorks,MapR等数十家公司基于开源的Hadoop平台构建自己的商业模式,可以认为是最近十年来最成功的开源社区。

Hadoop的成功固然是由于其顺应了新世纪以来互联网技术的发展趋势,同时其基于JVM的平台开发也为Hadoop的快速发展起到了促进作用。Hadoop生态圈的项目大都基于Java,Scala,Clojure等JVM语言开发,这些语言良好的语法规范,丰富的第三方类库以及完善的工具支持,为Hadoop这样的超大型项目提供了基础支撑。同时,作为在程序员中普及率最高的语言之一,它也降低了更多程序员使用,或是参与开发Hadoop项目的门槛。同时,基于Scala开发的Spark,甚至因为项目的火热反过来极大的促进了Scala语言的推广。但是随着Hadoop平台的逐步发展,Hadoop生态圈的项目之间的竞争加剧,越来越多的Hadoop项目注意到了这些JVM语言的一些不足之处,希望通过更有效率的处理方式,提升分布式系统的执行效率与健壮性。本文主要以Spark和Flink项目为例,介绍Hadoop社区观察到的一些因为JVM语言的不足导致的问题,以及相应的解决方案与未来可能的发展方向。

注:本文假设读者对Java和Hadoop系统有基本了解。

背景

目前Hadoop生态圈共有MapReduce,Tez,Spark及Flink等分布式计算引擎,分布式计算引擎项目之间的竞争也相当激烈。MapReduce作为Hadoop平台的第一个分布式计算引擎,具有非常良好的可扩展性,Yahoo曾成功的搭建了上万台节点的MapReduce系统。但是MapReduce只支持Map和Reduce编程范式,使得复杂数据计算逻辑需要分割为多个Hadoop Job,而每个Hadoop Job都需要从HDFS读取数据,并将Job执行结果写回HDFS,所以会产生大量额外的IO开销,目前MapReduce正在逐渐被其他三个分布式计算引擎替代。Tez,Spark和Flink都支持图结构的分布式计算流,可在同一Job内支持任意复杂逻辑的计算流。Tez的抽象层次较低,用户不易直接使用,Spark与Flink都提供了抽象的分布式数据集以及可在数据集上使用的操作符,用户可以像操作Scala数据集合类似的方式在Spark/FLink中的操作分布式数据集,非常的容易上手,同时,Spark与Flink都在分布式计算引擎之上,提供了针对SQL,流处理,机器学习和图计算等特定数据处理领域的库。

随着各个项目的发展与日益成熟,通过改进分布式计算框架本身大幅提高性能的机会越来越少。同时,在当前数据中心的硬件配置中,采用了越来越多更先进的IO设备,例如SSD存储,10G甚至是40Gbps网络,IO带宽的提升非常明显,许多计算密集类型的工作负载的瓶颈已经取决于底层硬件系统的吞吐量,而不是传统上人们认为的IO带宽,而CPU和内存的利用效率,则很大程度上决定了底层硬件系统的吞吐量。所以越来越多的项目将眼光投向了JVM本身,希望通过解决JVM本身带来的一些问题,提高分布式系统的性能或是健壮性,从而增强自身的竞争力。

JVM本身作为一个各种类型应用执行的平台,其对Java对象的管理也是基于通用的处理策略,其垃圾回收器通过估算Java对象的生命周期对Java对象进行有效率的管理。针对不同类型的应用,用户可能需要针对该类型应用的特点,配置针对性的JVM参数更有效率的管理Java对象,从而提高性能。这种JVM调优的黑魔法需要用户对应用本身以及JVM的各参数有深入的了解,极大的提高了分布式计算平台的调优门槛(例如这篇文章中对Spark的调优Tuning Java Garbage Collection for Spark Applications)。然而类似Spark或是Flink的分布式计算框架,框架本身了解计算逻辑每个步骤的数据传输,相比于JVM垃圾回收器,其了解更多的Java对象生命周期,从而为更有效率的管理Java对象提供了可能。

JVM存在的问题

1. Java对象开销

相对于c/c++等更加接近底层的语言,Java对象的存储密度相对偏低,例如【1】,“abcd”这样简单的字符串在UTF-8编码中需要4个字节存储,但Java采用UTF-16编码存储字符串,需要8个字节存储“abcd”,同时Java对象还对象header等其他额外信息,一个4字节字符串对象,在Java中需要48字节的空间来存储。对于大部分的大数据应用,内存都是稀缺资源,更有效率的内存存储,则意味着CPU数据访问吞吐量更高,以及更少的磁盘落地可能。

2. 对象存储结构引发的cache miss

为了缓解CPU处理速度与内存访问速度的差距【2】,现代CPU数据访问一般都会有多级缓存。当从内存加载数据到缓存时,一般是以cache line为单位加载数据,所以当CPU访问的数据如果是在内存中连续存储的话,访问的效率会非常高。如果CPU要访问的数据不在当前缓存所有的cache line中,则需要从内存中加载对应的数据,这被称为一次cache miss。当cache miss非常高的时候,CPU大部分的时间都在等待数据加载,而不是真正的处理数据。Java对象并不是连续的存储在内存上,同时很多的Java数据结构的数据聚集性也不好,在Spark的性能调优中,经常能够观测到大量的cache miss。Java社区有个项目叫做Project Valhalla,可能会部分的解决这个问题,有兴趣的可以看看这儿OpenJDK: Valhalla。

3. 大数据的垃圾回收

Java的垃圾回收机制,一直让Java开发者又爱又恨,一方面它免去了开发者自己回收资源的步骤,提高了开发效率,减少了内存泄漏的可能,另一方面,垃圾回收也是Java应用的一颗不定时炸弹,有时秒级甚至是分钟级的垃圾回收极大的影响了Java应用的性能和可用性。在当前的数据中心中,大容量的内存得到了广泛的应用,甚至出现了单台机器配置TB内存的情况,同时,大数据分析通常会遍历整个源数据集,对数据进行转换,清洗,处理等步骤。在这个过程中,会产生海量的Java对象,JVM的垃圾回收执行效率对性能有很大影响。通过JVM参数调优提高垃圾回收效率需要用户对应用和分布式计算框架以及JVM的各参数有深入的了解,而且有时候这也远远不够。

4. OOM问题

OutOfMemoryError是分布式计算框架经常会遇到的问题,当JVM中所有对象大小超过分配给JVM的内存大小时,就会fOutOfMemoryError错误,JVM崩溃,分布式框架的健壮性和性能都会受到影响。通过JVM管理内存,同时试图解决OOM问题的应用,通常都需要检查Java对象的大小,并在某些存储Java对象特别多的数据结构中设置阈值进行控制。但是JVM并没有提供官方的检查Java对象大小的工具,第三方的工具类库可能无法准确通用的确定Java对象的大小【6】。侵入式的阈值检查也会为分布式计算框架的实现增加很多额外的业务逻辑无关的代码。

解决方案

为了解决以上提到的问题,高性能分布式计算框架通常需要以下技术:

  1. 定制的序列化工具。显式内存管理的前提步骤就是序列化,将Java对象序列化成二进制数据存储在内存上(on heap或是off-heap)。通用的序列化框架,如Java默认的java.io.Serializable将Java对象以及其成员变量的所有元信息作为其序列化数据的一部分,序列化后的数据包含了所有反序列化所需的信息。这在某些场景中十分必要,但是对于Spark或是Flink这样的分布式计算框架来说,这些元数据信息可能是冗余数据。定制的序列化框架,如Hadoop的org.apache.hadoop.io.Writable,需要用户实现该接口,并自定义类的序列化和反序列化方法。这种方式效率最高,但需要用户额外的工作,不够友好。

  2. 显式的内存管理。一般通用的做法是批量申请和释放内存,每个JVM实例有一个统一的内存管理器,所有的内存的申请和释放都通过该内存管理器进行。这可以避免常见的内存碎片问题,同时由于数据以二进制的方式存储,可以大大减轻垃圾回收的压力。

  3. 缓存友好的数据结构和算法。只将操作相关的数据连续存储,可以最大化的利用L1/L2/L3缓存,减少Cache miss的概率,提升CPU计算的吞吐量。以排序为例,由于排序的主要操作是对Key进行对比,如果将所有排序数据的Key与Value分开,对Key连续存储,则访问Key时的Cache命中率会大大提高。
定制的序列化工具

分布式计算框架可以使用定制序列化工具的前提是要处理的数据流通常是同一类型,由于数据集对象的类型固定,对于数据集可以只保存一份对象Schema信息,节省大量的存储空间。同时,对于固定大小的类型,也可通过固定的偏移位置存取。当我们需要访问某个对象成员变量的时候,通过定制的序列化工具,并不需要反序列化整个Java对象,而是可以直接通过偏移量,只是反序列化特定的对象成员变量。如果对象的成员变量较多时,能够大大减少Java对象的创建开销,以及内存数据的拷贝大小。Spark与Flink数据集都支持任意Java或是Scala类型,通过自动生成定制序列化工具,Spark与Flink既保证了API接口对用户的友好度(不用像Hadoop那样数据类型需要继承实现org.apache.hadoop.io.Writable接口),同时也达到了和Hadoop类似的序列化效率。

Spark的序列化框架

Spark支持通用的计算框架,如Java Serialization和Kryo。其缺点之前也略有论述,总结如下:

占用较多内存。Kryo相对于Java Serialization更高,它支持一种类型到Integer的映射机制,序列化时用Integer代替类型信息,但还不及定制的序列化工具效率。 反序列化时,必须反序列化整个Java对象。 无法直接操作序列化后的二进制数据。 Project Tungsten 提供了一种更好的解决方式,针对于DataFrame API(Spark针对结构化数据的类SQL分析API,参考Spark DataFrame Blog),由于其数据集是有固定Schema的Tuple(可大概类比为数据库中的行),序列化是针对每个Tuple存储其类型信息以及其成员的类型信息是非常浪费内存的,对于Spark来说,Tuple类型信息是全局可知的,所以其定制的序列化工具只存储Tuple的数据,如下图所示

QQ截图20160714171128.jpg

图1 Spark off-heap object layout

对于固定大小的成员,如int,long等,其按照偏移量直接内联存储。对于变长的成员,如String,其存储一个指针,指向真正的数据存储位置,并在数据存储开始处存储其长度。通过这种存储方式,保证了在反序列化时,当只需访问某一个成员时,只需根据偏移量反序列化这个成员,并不需要反序列化整个Tuple。

Project Tungsten的定制序列化工具应用在Sort,HashTable,Shuffle等很多对Spark性能影响最大的地方。比如在Shuffle阶段,定制序列化工具不仅提升了序列化的性能,而且减少了网络传输的数据量,根据DataBricks的Blog介绍,相对于Kryo,Shuffle800万复杂Tuple数据时,其性能至少提高2倍以上。此外,Project Tungsten也计划通过Code generation技术,自动生成序列化代码,将定制序列化工具推广到Spark Core层,从而使得更多的Spark应用受惠于此优化。

Flink的序列化框架

Flink在系统设计之初,就借鉴了很多传统RDBMS的设计,其中之一就是对数据集的类型信息进行分析,对于特定Schema的数据集的处理过程,进行类似RDBMS执行计划优化的优化。同时,数据集的类型信息也可以用来设计定制的序列化工具。和Spark类似,Flink支持任意的Java或是Scala类型,Flink通过Java Reflection框架分析基于Java的Flink程序UDF(User Define Function)的返回类型的类型信息,通过Scala Compiler分析基于Scala的Flink程序UDF的返回类型的类型信息。类型信息由TypeInformation类表示,这个类有诸多具体实现类,例如(更多详情参考Flink官方博客Apache Flink: Juggling with Bits and Bytes):

  • BasicTypeInfo: 任意Java基本类型(装包或未装包)和String类型。

  • BasicArrayTypeInfo: 任意Java基本类型数组(装包或未装包)和String数组。

  • WritableTypeInfo: 任意Hadoop’s Writable接口的实现类.

  • TupleTypeInfo: 任意的Flink tuple类型(支持Tuple1 to Tuple25). Flink tuples是固定长度固定类型的Java Tuple实现。

  • CaseClassTypeInfo: 任意的 Scala CaseClass(包括 Scala tuples).

  • PojoTypeInfo: 任意的POJO (Java or Scala),例如,Java对象的所有成员变量,要么是public修饰符定义,要么有getter/setter方法。

    1. GenericTypeInfo: 任意无法匹配之前几种类型的类。)

前6种类型数据集几乎覆盖了绝大部分的Flink程序,针对前6种类型数据集,Flink皆可以自动生成对应的TypeSerializer定制序列化工具,非常有效率的对数据集进行序列化和反序列化。对于第7中类型,Flink使用Kryo进行序列化和反序列化。此外,对于可被用作Key的类型,Flink还同时自动生成TypeComparator,用来辅助直接对序列化后的二进制数据直接进行compare,hash等之类的操作。对于Tuple,CaseClass,Pojo等组合类型,Flink自动生成的TypeSerializer,TypeComparator同样是组合的,并把其成员的序列化/反序列化代理给其成员对应的TypeSerializer,TypeComparator,如下图所示:

QQ截图20160714171406.jpg

图2 Flink组合类型序列化

此外,如有需要,用户可通过集成TypeInformation接口,定制实现自己的序列化工具。

显式的内存管理

垃圾回收的JVM内存管理回避不了的问题,JDK8的G1算法改善了JVM垃圾回收的效率和可用范围,但对于大数据处理的实际环境中,还是远远不够。这也和现在分布式框架的发展趋势有冲突,越来越多的分布式计算框架希望尽可能多的将待处理的数据集放在内存中,而对于JVM垃圾回收来说,内存中Java对象越少,存活时间越短,其效率越高。通过JVM进行内存管理的话,OutOfMemoryError也是一个很难解决的问题。同时,在JVM内存管理中,Java对象有潜在的碎片化存储问题(Java对象所有信息可能不是在内存中连续存储),也有可能在所有Java对象大小没有超过JVM分配内存时,出现OutOfMemoryError问题。

Flink的内存管理

Flink将内存分为三个部分,每个部分都有不同的用途:

  • Network buffers: 一些以32KB Byte数组为单位的buffer,主要被网络模块用于数据的网络传输。

  • Memory Manager pool: 大量以32KB Byte数组为单位的内存池,所有的运行时算法(例如Sort/Shuffle/Join)都从这个内存池申请内存,并将序列化后的数据存储其中,结束后释放回内存池。

  • Remaining (Free) Heap: 主要留给UDF中用户自己创建的Java对象,由JVM管理。

Network buffers在Flink中主要基于Netty的网络传输,无需多讲。Remaining Heap用于UDF中用户自己创建的Java对象,在UDF中,用户通常是流式的处理数据,并不需要很多内存,同时Flink也不鼓励用户在UDF中缓存很多数据,因为这会引起前面提到的诸多问题。Memory Manager pool(以后以内存池代指)通常会配置为最大的一块内存,接下来会详细介绍。

在Flink中,内存池由多个MemorySegment组成,每个MemorySegment代表一块连续的内存,底层存储是byte[],默认32KB大小。MemorySegment提供了根据偏移量访问数据的各种方法,如get/put int,long,float,double等,MemorySegment之间数据拷贝等方法,和java.nio.ByteBuffer类似。对于Flink的数据结构,通常包括多个向内存池申请的MemeorySegment,所有要存入的对象,通过TypeSerializer序列化之后,将二进制数据存储在MemorySegment中,在取出时,通过TypeSerializer反序列化。数据结构通过MemorySegment提供的set/get方法访问具体的二进制数据。

Flink这种看起来比较复杂的内存管理方式带来的好处主要有:

  • 二进制的数据存储大大提高了数据存储密度,节省了存储空间。

  • 所有的运行时数据结构和算法只能通过内存池申请内存,保证了其使用的内存大小是固定的,不会因为运行时数据结构和算法而发生OOM。而对于大部分的分布式计算框架来说,这部分由于要缓存大量数据,是最有可能导致OOM的地方。

  • 内存池虽然占据了大部分内存,但其中的MemorySegment容量较大(默认32KB),所以内存池中的Java对象其实很少,而且一直被内存池引用,所有在垃圾回收时很快进入持久代,大大减轻了JVM垃圾回收的压力。

  • Remaining Heap的内存虽然由JVM管理,但是由于其主要用来存储用户处理的流式数据,生命周期非常短,速度很快的Minor GC就会全部回收掉,一般不会触发Full GC。

Flink当前的内存管理在最底层是基于byte[],所以数据最终还是on-heap,最近Flink增加了off-heap的内存管理支持,将会在下一个release中正式出现。Flink off-heap的内存管理相对于on-heap的优点主要在于(更多细节,请参考Apache Flink: Off-heap Memory in Apache Flink and the curious JIT compiler):

  • 启动分配了大内存(例如100G)的JVM很耗费时间,垃圾回收也很慢。如果采用off-heap,剩下的Network buffer和Remaining heap都会很小,垃圾回收也不用考虑MemorySegment中的Java对象了。

  • 更有效率的IO操作。在off-heap下,将MemorySegment写到磁盘或是网络,可以支持zeor-copy技术,而on-heap的话,则至少需要一次内存拷贝。

  • off-heap可用于错误恢复,比如JVM崩溃,在on-heap时,数据也随之丢失,但在off-heap下,off-heap的数据可能还在。此外,off-heap上的数据还可以和其他程序共享。
Spark的内存管理

Spark的off-heap内存管理与Flink off-heap模式比较相似,也是通过Java UnSafe API直接访问off-heap内存,通过定制的序列化工具将序列化后的二进制数据存储与off-heap上,Spark的数据结构和算法直接访问和操作在off-heap上的二进制数据。Project Tungsten是一个正在进行中的项目,想了解具体进展可以访问:[SPARK-7075] Project Tungsten (Spark 1.5 Phase 1), [SPARK-9697] Project Tungsten (Spark 1.6)。

缓存友好的计算

磁盘IO和网络IO之前一直被认为是Hadoop系统的瓶颈,但是随着Spark,Flink等新一代的分布式计算框架的发展,越来越多的趋势使得CPU/Memory逐渐成为瓶颈,这些趋势包括:

  • 更先进的IO硬件逐渐普及。10GB网络和SSD硬盘等已经被越来越多的数据中心使用。

  • 更高效的存储格式。Parquet,ORC等列式存储被越来越多的Hadoop项目支持,其非常高效的压缩性能大大减少了落地存储的数据量。

  • 更高效的执行计划。例如Spark DataFrame的执行计划优化器的Fliter-Push-Down优化会将过滤条件尽可能的提前,甚至提前到Parquet的数据访问层,使得在很多实际的工作负载中,并不需要很多的磁盘IO。

由于CPU处理速度和内存访问速度的差距,提升CPU的处理效率的关键在于最大化的利用L1/L2/L3/Memory,减少任何不必要的Cache miss。定制的序列化工具给Spark和Flink提供了可能,通过定制的序列化工具,Spark和Flink访问的二进制数据本身,因为占用内存较小,存储密度比较大,而且还可以在设计数据结构和算法时,尽量连续存储,减少内存碎片化对Cache命中率的影响,甚至更进一步,Spark与Flink可以将需要操作的部分数据(如排序时的Key)连续存储,而将其他部分的数据存储在其他地方,从而最大可能的提升Cache命中的概率。

Flink中的数据结构

以Flink中的排序为例,排序通常是分布式计算框架中一个非常重的操作,Flink通过特殊设计的排序算法,获得了非常好了性能,其排序算法的实现如下:

  • 将待排序的数据经过序列化后存储在两个不同的MemorySegment集中。数据全部的序列化值存放于其中一个MemorySegment集中。数据序列化后的Key和指向第一个MemorySegment集中其值的指针存放于第二个MemorySegment集中。

  • 对第二个MemorySegment集中的Key进行排序,如需交换Key位置,只需交换对应的Key+Pointer的位置,第一个MemorySegment集中的数据无需改变。 当比较两个Key大小时,TypeComparator提供了直接基于二进制数据的对比方法,无需反序列化任何数据。

  • 排序完成后,访问数据时,按照第二个MemorySegment集中Key的顺序访问,并通过Pinter值找到数据在第一个MemorySegment集中的位置,通过TypeSerializer反序列化成Java对象返回。

QQ截图20160714171841.jpg

图3 Flink排序算法

这样实现的好处有:

  1. 通过Key和Full data分离存储的方式,尽量将被操作的数据最小化,提高Cache命中的概率,从而提高CPU的吞吐量。

  2. 移动数据时,只需移动Key+Pointer,而无须移动数据本身,大大减少了内存拷贝的数据量。

  3. TypeComparator直接基于二进制数据进行操作,节省了反序列化的时间。
Spark的数据结构

Spark中基于off-heap的排序与Flink几乎一模一样,在这里就不多做介绍了,感兴趣的话,请参考:https://databricks.com/blog/2015/04/28/project-tungsten-bringing-spark-closer-to-bare-metal.html

总结

本文主要介绍了Hadoop生态圈的一些项目遇到的一些因为JVM内存管理导致的问题,以及社区是如何应对的。基本上,以内存为中心的分布式计算框架,大都开始了部分脱离JVM,走上了自己管理内存的路线,Project Tungsten甚至更进一步,提出了通过LLVM,将部分逻辑编译成本地代码,从而更加深入的挖掘SIMD等CPU潜力。此外,除了Spark,Flink这样的分布式计算框架,HBase(HBASE-11425),HDFS(HDFS-7844)等项目也在部分性能相关的模块通过自己管理内存来规避JVM的一些缺陷,同时提升性能。