当前NoSQL的分布式存储系统的资料(包括Paper以及各大Blog上的内容),大部分都是关于分布式Key-Value存储下的读写操作、数据一致性、节点失效与修复等问题上,很少有系统的提及有关Load Balance的处理。不过既NoSQL指导型原则是CAP原理,而系统在实现时无论偏重AP还是CP,P都是必选的,因此对于分布式的各个节点的负载均衡也应该是要面对的问题,不过很遗憾,在BigTable, Dynamo以及Cassandra的Paper中对于Load Balance基本都没有提及,都只是很简单的进行了说明。特地花了几天去找了一些资料,总结记录于此。

1. Dynamo

Dynamo的论文上对于Load Balance的提及基本就是一个关键词”uniform hashing”,”A uniform key distribution can help us achieve uniform load distribution assuming the access distribution of keys is not highly skew“。即通过均匀的hash函数,保证将所有的key以及node都可以均匀的分布到同一哈希地址空间上,从而实现负载的均衡,此类常用的函数包括MD5 hash函数和SHA-1 hash函数。

除了利用哈希函数使得整个存储的item被均匀的分配到整个地址空间上,此外Dynamo的Virtual Node机制也对负载均衡进行一定的帮助,根据机器的性能指标不同划分不同的VNode即可。

此外,当新的VNode启动以及离开时,需要将某个VNode的数据迁移至邻居节点,而此种迁移有时会带来比较大的计算量(Merkle Tree等),因此Dynamo又采用了一个将整个地址空间划分成段的策略(如下图所示),然后每个VNode只能在一个段边界位置进行添加,即每一个VNode管理段长的整数倍的地址空间,此种方式对于节点的启动以及恢复,数据的归档等都有好处。

由于Dynamo虽然是NoSQL数据库的雏形以及类white paper,但由于工业界基于此模型构建实际针对业务数据库并是特别认可,因此资料也较少。

2.  Cassandra

Cassandra首先包含了两种Partitioner,这两种Partitioner对负载均衡的影响完全不同。

RandomPartitioner: RandomPartitioner采用均匀的hash函数,将负载均匀的划分到Key space空间上。RandomPartitioner gives you pretty good load balancing with no further work required。不过由于hash函数会将连续的Key映射到不同的环上地址,缺点就是无法进行range query.
OrderPreservingPartitioner:OrderPreservingPartitioner会使得连续的key映射到环上的地址空间后也是连续的,因此支持range query. OrderPreservingPartitioner let u perform range queries on the keys, but requires choosing node tokens carefully and active load balancing.

也就是说RandomPartitioner几乎是不需要进行Load Balancing的,但对于后者,就需要在Load Balancing上下一点功夫。

Cassandra的负载均衡,提供了一种在哈希的地址空间上Move Node的方式,不过本质上这种节点Move的行为还是基于节点的“退出+重新在另一位置启动”的操作的。Cassandra开源的nodetools ring中提供rebalance的功能,具体的选择可以有两种方式。1. Autobootstrap + Automatic Token Algorithm: 每次有节点启动的时候,分配给节点一个可以分担负载最重节点1/2的位置(Assign a token that will give me half of the range of the most loaded node)。2. nodeltool move + Manual token assignment:  cassandra的wiki有一段python的计算位置的代码(其实就是平均分割地址空间)。有关这个的描述可以查看http://www.datastax.com/blog/slides-and-videos-cassandra-summit-2010

另外就是既然Cassandra是开源的,那直接研究代码,找到org.apache.cassandra.service.StorageLoadBalancer.java,前面的注释中已经给了我们一个很大的参考,可以看到Cassandra的Load Balance算法是基于一篇PDCS05的Paper的,对此有兴趣的可以参看<Scalable range query  processing for large-scale distributed database applications>

Load balance的核心算法就是根据设置阈值,设L是平均负载,Li是节点i的负载,ε是系数。

Case 1: 当L≤Li≤εL时,认为节点处于适载,不进行Load Balance操作。

Case 2: 当Li > εL,节点超载,则分成以下两种情况:  a. 当i节点的前驱节点与后继节点中最小者j节点,使得满足 (Li + Lj)/2 ≤ εL时,则直接利用邻居节点进行Load Balancing操作,如果较轻的是前驱节点,则前驱节点顺时针方向移动,后继则就是i节点自身逆时针移动,使得负载迁移到后继节点。迁移量都是(Li-Lj)/2,效果就是使得两个i,j节点平均分配负载,负载后都为(Li+Lj)/2; b. 当前驱后继节点无法满足条件时,则需要从系统中随机找到一个underloaded节点(在下面描述)来从系统中寻找到一个可以move的underloaded节点进行位置迁移来分担i的负载。

Case3:当Li < L时,节点轻载(underloaded),此时也分两种情况。a. 如果(Li + L(i+1)) > εL,表示该节点不可以迁移走,就什么事情都不做。 b 如果(Li + L(i+1)) ≤ εL,表示该节点可以迁移并且其后继不会overloaded,因此它就像系统中随机一个节点通知自己是underloaded,以等待有某个overloaded节点在系统中请求。

下面就是从StroageLoadBalancer.java中摘取的有关该算法的几个函数。

/* If a node's load is this factor more than the average, it is considered Heavy */
/* 虽然在其他地方看到了有关这个系数有说2是最好的,但cassandra的源代码中还是采用了1.5 */
    private static final double TOPHEAVY_RATIO = 1.5;
/* 计算系统中的平均负载  */
    private double averageSystemLoad()
    {
        int nodeCount = loadInfo2_.size();
        Set<InetAddress> nodes = loadInfo2_.keySet();

        double systemLoad = 0;
        for (InetAddress node : nodes)
        {
            systemLoad += loadInfo2_.get(node);
        }
        double averageLoad = (nodeCount > 0) ? (systemLoad / nodeCount) : 0;
        if (logger_.isDebugEnabled())
            logger_.debug("Average system load is {}", averageLoad);
        return averageLoad;
    }
/* 是否超载  */
     private boolean isHeavyNode()
    {
        return ( localLoad() > ( StorageLoadBalancer.TOPHEAVY_RATIO * averageSystemLoad() ) );
    }
/* 是否可移动, Case 3的情况  */
    private boolean isMoveable(InetAddress target)
    {
        double threshold = StorageLoadBalancer.TOPHEAVY_RATIO * averageSystemLoad();
        if (isANeighbour(target))
        {
            // If the target is a neighbour then it is
            // moveable if its
            Double load = loadInfo2_.get(target);
            if (load == null)
            {
                return false;
            }
            else
            {
                double myload = localLoad();
                double avgLoad = (load + myload) / 2;
                return avgLoad <= threshold;
            }
        }
        else
        {
            InetAddress successor = StorageService.instance.getSuccessor(target);
            double sLoad = loadInfo2_.get(successor);
            double targetLoad = loadInfo2_.get(target);
            return (sLoad + targetLoad) <= threshold;
        }
    }

3. Hypertable

Hypertable是一款基于Bigtable论文的开源实现,Bigtable是王道但国内公司做不起,开源的Hypertable&HBase这两年应该可以做线下及半线下应用,类似Bigtable的方案做线上服务开源界很难搞定(摘自http://www.nosqlnotes.net/archives/128),想起来查找这个的资料主要是由于bigtable中有关负载均衡的描述实在寥寥无几,幸好的是Hypertable还些有点内容。

有关Hypertable的简单介绍可以看下这里,特别是对于range, rangeserver的理解有助于对于负载均衡原理以及过程的理解。

Hypertable负载均衡操作一般会由一下三种情况触发。

1. Load Imbalance:

Hypertable的日常balance操作每天执行一次,每天具体的执行时间可以根据上一周的server load map进行决定,添加在那个时间的balance task。而balance的操作仅参考之前24小时的server load数据,统计24小时每台server的平均load map(loadavg),并采用配置的偏差量的threshold来决定是否对某台server进行balance,任何时候当有新节点加入后,系统会至少等待24小时才进行load balance操作以使得新加入的节点能够构建24小时的load统计量。有关这个Threshold的设定需要稍微大一点,至少要容忍Range因为增长分裂后转移新分裂Range给其他Server所产生的负载变化。

操作过程如下:

a. 建立一个server_load_vec(vector),vector的每项内容为<loadavg, derivation, server_name>,loadavg是server过去24小时的average load,而derivation则是自身的loadavg与整个系统中所有server的loadavg的average的偏差。

b. 均衡算法每次减轻一台server的负载,当处理到某台server时,构建server上range的负载描述vector, 称为range_load_vec,每项内容为<loadestimate, start_row, end_row>,而对于任何最近一周内才加入或者迁移到本机器的range则不统计。其中loadestimate 由该range每秒的操作(scan/update, cell r/w, byte r/w等因素决定),并计算一个所有server loadavg在loadestimate上的平均值loadavg_per_loadestimate = loadavg/sum(loadestimate).

算法从偏差最大的的开始处理,并迁移range,执行过程伪代码如下:

server_load_vec_desc = sort_descending(server_load_vec);
server_load_vec_asc = sort_ascending(server_load_vec);
while (server_load_vec_desc[0].deviation > DEVIATION_THRESHOLD) {
  populate_range_load_vector(server_load_vec_desc[0].server_name);
  sort descending range_load_vec;
  i=0;
  while (server_load_vec_desc[0].deviation > DEVIATION_THRESHOLD &&
            i < range_load_vec.size()) {
    if (moving range_load_vec[i] from server_load_vec_desc[0] to server_load_vec_asc[0] reduces deviation) {
       add range_load_vec[i] to balance plan
       partial_deviation = range_load_vec[i].loadestimate * loadavg_per_loadestimate;
       server_load_vec_desc[0].loadavg -= partial_deviation;
       server_load_vec_desc[0].deviation -= partial_deviation;
       server_load_vec_asc[0].loadavg += partial_deviation;
       server_load_vec_asc[0].deviation += partial_deviation;
       server_load_vec_asc = sort_ascending(server_load_vec_asc);
    }
    i++;
  }
  if (i == range_load_vec.size())
    remove server_load_vec_desc[0] and corresponding entry in server_load_vec_asc  
  server_load_vec_desc = sort_descending(server_load_vec_desc);
}

2. 添加一个RangeServer:

一个RangeServer被添加进入系统后,立刻进行load balancing操作,不过此处的balancing的算法采用一种很粗鲁和随意(slight)的方式,即随机的的从其他servers中选则range迁移到新的RangeServer,并不会去寻找负载最重的RangeServer去进行均衡操作。

3. 对一个range进行split操作生成新range

当一个range由于数据的添加以及修改等操作导致数据过大需要分裂时,采用的方式是将该range所属的table的所有ranges尽可能的平均分配到各个RangeServer上,即Balancer收集所有RangeServer上与该range所属的table的ranges的信息,统计各个RangeServer上存储该table的range的个数,存储该table的range数最小的那台server将会被分配去存储新分裂出来的range.

转载请注明来源:Leoncom-《Dynamo,Cassandra,Hypertable的负载均衡策略》
, , , ,
Trackback

only 1 comment untill now

  1. Most Nigerians want to make money quickly without any work to be
    done and this is why so many are being scammed and duped. In publishing your ebook,
    make sure that people will not have a hard time to download your ebook internet marketing strategy and it is easy to locate.

    Similarly, you can write guest posts on established blogs
    and contribute to other blogs through.

Add your comment now