SQL-on-Hadoop摘要

原始文章传送门 http://www.gruter.com/blog/?p=391

General Feature
数据往往以RC & ORC & Parquet等格式存放在HDFS 以及在分布式环境上执行SQL

SQL-on-Hadoop solutions
1.  Hive: 将SQL转成MapReduce Job,继承了MR的稳定可靠,同时也带来了Shuffle过程的额外消耗
2.  Stinger: 将计算框架迁移到Tez上, 因此并不算是一个独立的新解决方案,但可以解决Hive一部分不足
3.  Apache Iajo: 在HDFS上使用了自己的分布式执行引擎
4.  Impala: 使用了自己的分布式计算框架,在Shuffer阶段会将所有数据保存在内存中,因此避免了shuffling overheads,  但是不能处理超过内存限制的数据量
5.  Apache Drill: Google Dremel的开源实现,目前还在开发中,而且还没实现其主要的分布式处理模块

Key SQL-on-Hadoop performance factors
1.  磁盘扫描速度,因为数据通常是存放在HDFS;此外中间数据的传输效率也是个重要因素,因为像Group By或者Order By之类的算子往往会以多步来执行
2.  Query Execution启动时间,对于短查询而言,启动时间会成为较大的OverHeads,而在长查询中,启动时间往往占比很小
3. 文件格式,目前有 RC & ORC & Parquet等,往往这些文件格式有不同的特性,往往需要根据实际的应用场景来选择合适的文件。Query往往会生成大量的结果文件,文件格式和文件写效率是一个关键的因素,比如ORC文件格式有较好的读性能,但是写性能较差。此外合适的压缩机制也是一个关键因素在SQL-on-Hadoop系统中

The absolute performance limits of SQL-on-Hadoop
1.  除了上面的因素之外,还有一些限制是不能超出的:处理性能会小于或者等于在HDFS上的磁盘带宽
2.  对于像Select count(*) from table这样的查询而言,由于没有shuffle消耗,因此磁盘读取速度,Query启动时间和文件格式成为关键因素,当使用相同的文件格式和大量的数据场景下,各个解决方案的时间相差不远
3. 对于across a range of natural field queries,  Stinger这些只是比Hive快1.5-3倍左右

Frequently-used performance comparison queries
1.  对于短查询而言,Stinger由于省去了启动时间,因此可能会比Hive快上百倍,但是并不意味着在长查询中也有同样的效果
2.  而对于Order By这类操作,Hive只用单机来执行,对于能够分布式Order By方案而言,查询效率是能够比Hive高上千倍的,只要节点数足够多。但是并不能就宣称比Hive快上上千倍

Scaling Memcache at Facebook学习笔记

基本情况:

  1. the largest memcached installation in the world
  2. provides a suite of configuration, aggregation and routing services to organize memcached instances into a distributed system
  3. 使用旁路型cache-demand-filled look-aside cache
  4. 在不同的Scale需要解决的问题重点不一样:performace efficiency fault-tolerance consistency

基本操作

  1. Query Cache:先从memcache读取,然后出现miss,再从DB读取,最后写入cache
  2. Update Cache:  先update DB 然后在cache中删除。采用delete cache操作,是因为delete是幂等操作,可以防止出现并发写造成的数据不一致现象

单集群场景

  1. 一个集群内部有thousands of server, 面临的主要问题有两个:读取cache的延迟以及cache miss造成的负载
  2. 解决延迟主要的方式主要在Client端的优化,Client主要负责了序列化、压缩等操作,同时会通过Configuration System更新下游可用的Cache Server
  3. 延迟优化方案-并行化和批量请求:会根据数据之间的依赖关系构造一个DAG,根据DAG最大化可以并行请求的数据数。平均一次请求会有24个key
  4. 延迟优化方案-不同场景分别使用UDP和TCP。对于读请求使用了UDP,前端每个Client的每个线程都直接与memcache通信。Client端在读取失败时会认为是cache miss,但是在这种场景下的cache miss并不会把从DB读取的结果再写入cache,这是为了避免给整个集群的网络带来额外负担。对于更新请求使用了TCP,Client端通过一个部署在同一机器上的mcrouter与memcache交互,引入中间层可以减少Client每个线程维护大量TCP连接带来的网络、内存和CPU消耗。按照论文而言,relying on UDP can lead to a 20% reduction in latency to server requests
  5. 延迟优化方案-使用滑动窗口的流控机制来减少incast congestion。当一个Client发出了大量读请求,如果这些请求的response同时到达,将会对机架和集群交换机的带宽造成了过载压力。滑动窗口过小会造成不必要的延迟,窗口过大会造成incast congestion,导致落到后端DB的压力变大,进一步延迟会变大
  6. 负载优化方案-采用Lease机制。Lease是一个64-bit的Token,由MemCache在读请求时分配给Client,可以用来解决两个问题:stale sets 和thundering herds。stale sets是指在Query Cache和Update Cache并发执行时,可能会导致Query Cache在Cache Miss的情况下会往cache中写入老数据,通过在写入时加Lease验证+Lease会在接收到delete操作时会失效的机制可以解决这个问题。thundering herds发生在一些特殊的key写操作频繁同时很多并发读请求。当大量读请求出现cache miss时就会导致后端压力过大。Cache可以在原有的lease机制做些小改进,并不是每个读请求都分配Lease,而是每个key 10s才分配一次。那么Client在没有获取到lease时 可以稍等一段时间再重新访问cache,此时cache中已有数据。通过这样的方式,可以使得这些特殊的key的DB读请求由17K/s减少到1.3K/s。此外还有个Stale values的tricky技巧,当key接收到删除操作时,value会放到一个最近删除列表,等待过一段短时间才会真正删除(类似于延迟删除)然后部分应用可以接受读取这些待删除的不一致的脏数据。
  7. 负载优化方案-拆分成Memcache Pools。如果不同的应用共享同一个Cache集群,那么使用cache的场景、内存和负载等都会对cache命中率有消极作用。因此为了适应这种差异性,将会把一个集群的cache server拆分成不同的Pool。比如会对一些访问频率低但是cache miss代价很高的key分配large pool。
  8. 负载优化方案-Memcache Pools的多副本机制。对于并发访问key多,数据集不超出一两台cache server的容量范围但是请求数超过了单机处理能力的Pools,会将cache数据作replication。而主要的问题就在于如何报纸多副本的数据一致性。
  9. 负载优化方案-容错处理。对于小规模的cache访问失败(比如由于网络原因或者server宕机),引入了Gutter Pool机制。Gutter是一个小规模cache集群,大致占集群的1%机器。在第一次正常访问cache失败后,会继续访问Gutter,如果再次访问失败,将会把从DB读取到的数据填入到Gutter。Gutter Pool里面的数据过期时间较短,因此没有delete操作,带来的问题就是会读取短期的脏数据。而直接把key rehash到其他剩余存活机器的方法相比于Gutter Pool方式,会存在雪崩的风险。因为一些热门key被rehash会会增加当前机器的负载,可能会导致过载。而Gutter Pool一般负载较低,可以降低这种风险。

多集群场景-In a region

  1. Split web and memcached servers into multiple frontend clusters. These clusters, along with a storage cluster that contain the databases, define a region. 主要考虑的问题是如何在同一个region内部多个集群之间备份数据
  2. Regional Invalidations-在一个region内部会让storage cluster负责cache数据失效的功能。每个DB上会部署一个mcsqueal,负责从commit log里面提取出删除操作,然后发送到前端。mcsqueal并不是直接把删除操作广播到所有Cache Server,而是先转发到一些特定机器上的mcrouter,由mcrouter根据每个key转发到合适的Cache Server上,这是为了减少集群网络包的个数以及网络压力。而相比于直接由web server直接广播删除操作的方式,一来是可以更好地packet delete操作,二来是通过log的方式可以更方便的回放操作,这可以解决delete操作发送失败以及发错到其它cache Server的问题。此外还存在一个优化手段,一个web server修改数据后会往本集群的cache发送delete操作,这是为了能够给客户端提供一个read-after-write语义(即修改后需要读取新数据)以及减少脏数据在本地存活的时间。
  3. Regional Pools-如果落在前端的请求是随机转发,那么所有cache集群保存的key都基本一致,这种数据副本过多会造成内存的极大浪费。为了减少不必要的数据副本数,可以将多个前端集群共享一批cache Server的方式,这个称之为Regional Pools。数据备份是用更多的机器来换取更少的带宽使用、更少的延迟和更好的容错。而Regional Pools是希牺牲了数据备份的优势,在Regional Pools内部跨集群的通信延迟会变大,因此决定哪些key需要放到Regional Pools需要考虑这些因素:访问频率,数据大小和独立用户访问数,一般访问频率低、数据量小的数据会放到Regional Pools。
  4. Cold Cluster Warm Up-新加入的cache集群由于cache命中率较低,为了避免cache miss对后端DB压力过大,可以允许从有正常命中率的cache集群(Warm Cluster)读取数据.需要注意可能会出现读到脏数据的场景,这是因为delete操作没有及时同步到Warm Cluster,会出现从Warm Cluster读取到旧数据然后再写入到Cold Cluster。可以采取延迟删除的机制来解决,对Cold Cluster的delete操作会延迟2s后才真正删除,在此之间任何add操作都会失败。虽然2s这个时间理论上仍然存在读取到脏数据的可能,但是实际上在大多数情况下是可行的。当Cold Cluster cache命中率正常后,将会关闭填充这种填充cache机制。

多集群场景-Across Region

  1. 多个数据中心情况下会建立多个regions. 其中一个region会放置主DB,而其他region提供只读的replica,主从之间通过mysql的同步机制来同步数据。在跨集群的场景下面临的主要问题是如何解决由于主从同步延迟造成的数据不一致问题。
  2. 主从同步延迟主要是可能Server A已经往主库写入新数据了,但Server B由于cache miss由从库中读到的还是老数据,那么填入cache的就是脏数据,可以使用remote marker机制来解决cache和DB数据不一致问题。在需要更新DB数据之前,先根据key生成一个remote marker r(key)并更新cache region中,这是用来标明从库的数据是脏数据。然后再按照正常的逻辑更新DB数据。而在读取数据时如果出现cache miss,那么会先检查是否存在r(key),如果有则从主库读取数据,否则由从库读取数据。当从库同步完数据会由mcsqueal在cache region中删除r(key)。
  3. 这种remote marker机制会存在一种风险,当多个Web Server出现并发写同一个key的时候可能会导致r(key)被过早删除。按照论文中注明,这种冲突场景极少出现

单机优化

  1. 性能优化-允许内部使用的hashtable自动扩张(rehash)、由原先多线程下的全局单一的锁改进为更细粒度的锁以及每个线程一个UDP来避免竞争
  2. 内存优化-改进Slab Allocator,允许不同Slab Class互换内存。当发现要删除的item对应的Slab Class最近使用的频率超过平均的20%,将会把最近最少使用的Slab Class的内存交互给需要的Slab Class使用。整体思路类似一个全局的LRU。
  3. Transient Item Cache-某些key的过期时间会设置的较短,如果这些key已经过期但是还没达到LRU尾部时,它所占用的内存在这段时间就会浪费。因此可以把这些key单独放在一个由链表组成的环形缓冲(即相当于一个循环数组,数组的每个元素是个链表,以过期时间作为数组的下标),每隔1s,会将缓冲头部的数据给删除。
  4. 软件升级-memcache的数据是保存在System V的共享内存区域,方便机器上的软件升级。