通过写代码学习AWS DynamoDB （3）- 一致性hash
简介
在本文中，我们将简单介绍一致性hash（consistent hash）的概念，以及一致性hash可以解决的问题。然后我们将在模拟的DDB实现中实现一个简单版本的基于一致性harsh实现的partition。
问题
在《通过写代码学习AWS DynamoDB （2）》中我们的DDB使用了最朴素的hash算法来分配一个key/value存储的partition。也就是使用
key % <partition count>的方法。当我们需要改变partition的数量来达到scale out和scale in的目的时，我们发现使用这种方法算出来的新的partition会和之前的partition有很大的区别。例如，如果我们把partition的数量从10增加到11时，新分配的partition和以前分配的partition会如下表所示：
| Key | Existing Partition | New Partition |
| 1567 | 7 | 5 |
| 2354 | 4 | 0 |
| 2888 | 8 | 6 |
| 8971 | 1 | 6 |
Key所对应的partition的改变意味着我们需要把该partition的数据转移到新的partition上。这是一种很大的开销。而在partition数量改变时，一致性hash可以帮助我们减少数据所对应的partition发生改变的次数。
一致性hash介绍
一致性hash的介绍文章在网上很多，大家可以自行参看。这里仅仅做一个简单的介绍。首先，我们介绍下面几个和hash有关的概念：
Hash空间：Hash空间是指全部可能的hash key组成的空间。比如说，我的hash key是用“x % 1000”产生的，那么我的hash空间就是0到999，因为所有可能的hash key都处在这个范围内，并且这个空间的每一个值也都可能是一个hash key。
Hash key：Hash key是用hash函数产生的key。对于每一个需要映射到hash空间的对象都需要使用hash函数产生出一个hash key。
Hash server：这里Hash server是指将一个server或者host通过某种方法映射到Hash空间。比如，使用IP地址来进行映射。
下面的图描述了一个一致性hash的实现原理。
图中的A，B，C是3个hash server，它们已经被映射到了hash 空间上。Kate，John，Steve，Bill，和Jane是5个element，它们也被映射到了hash空间上。现在我们规定每一个element将被保存到与它相邻的前面的hash server中。这样，John，Steve被保存到C中，Bill被保存到B中，Kate和Jane被保存到A中。通过这种方式，我们可以看到如果需要增加或者减少Hash server，理论上只会有1/N的元素被移动到新的hash server里，N是全部hash server的数量。而一般性的hash实现方法几乎是全部数据需要被移动，所以这是一个很大的改进。
这种普通的consistent hash的实现仍然具有一些缺点。例如当我们分配element给hash server时，可能会出现非常不均匀的情况。某些hash server可能会保存比其它hash server多很多的element。这是我们试图避免的。所以，又产生了Virtual node的概念。如下图所示：
现在映射到hash空间的都是每一个hash server的virtual node。每一个hash server可以有很多的virtual node。比如图中每一个hash server有3个virtual node。现在每一个hash空间里的element是被分配到邻近的virtual node上，而通过virtual node又可以找到实际的hash server。通过将每一个hash server映射出多个virtual node的方式，我们可以大大减小element分配不均匀的问题。
实现一致性hash
我们在这里实现的是最基础的一致性hash，但是基于virtual node的实现本身也不复杂很多，只是多了一层virtual node到hash server的mapping。首先，我们将partition的管理从DDB.Table里移出来，并且定义了一个类Consistent_hash_manager，由它管理partition。这个类不仅提供了对key/value的增删改操作，同时提供了scale_out_partition()和scale_in_partition()两个操作。通过这两个操作我们可以增加和删除partition。这样DDB.Table就不再需要知道partition的细节，而仅仅需要通过Consistent_hash_manager的接口直接对数据和partition进行操作。修改之后的DDB和Consistent_hash_manager如下所示：
DDB 类的实现：
from consistent_hash_manager import Consistent_hash_manager
# class to provide DDB public APIs
# - support partitions based on hash value of key;
class DDB:
def __init__(self):
self.tables = {}
def create_table(self, table_name):
self.tables[table_name] = self.Table(table_name)
def list_table(self):
for table in self.tables.values():
table.describe()
def delete_table(self, table_name):
self.tables.pop(table_name)
def get_table(self, table_name):
return self.tables[table_name]
class Table:
def __init__(self, name, partition_count=1):
self.name = name
self.consistent_hash_manager = Consistent_hash_manager(partition_count)
def put_item(self, key, value):
self.consistent_hash_manager.put_item(key, value)
def update_item(self, key, value):
self.consistent_hash_manager.put_item(key, value)
def get_item(self, key):
return self.consistent_hash_manager.get_item(key)
def delete_item(self, key):
self.consistent_hash_manager.delete_item(key)
def describe(self):
item_count = self.consistent_hash_manager.get_item_count()
print("Table name: {}, item size: {}".format(self.name, item_count))
def scale_out(self):
self.consistent_hash_manager.scale_out_partition()
def scale_in(self):
self.consistent_hash_manager.scale_in_partition()Consistent_hash_manager的实现：
from random import randrange
import functools
from partition import Partition
class Consistent_hash_manager:
def __init__(self, partition_count=3):
self.hash_space = 10
self.available_hash_keys = list(range(self.hash_space))
self.partitions = list()
for _ in range(partition_count):
self.partitions.append(Partition(self.available_hash_keys[randrange(len(self.available_hash_keys))]));
self.available_hash_keys.remove(self.partitions[-1].get_id())
# sort by id so partitions build a consistent hash ring
self.partitions.sort(key=lambda x : x.get_id())
# add one partition
def scale_out_partition(self):
if len(self.available_hash_keys) == 0:
print("No available hash space for scale out")
return
self.partitions.append(Partition(self.available_hash_keys[randrange(len(self.available_hash_keys))]))
self.available_hash_keys.remove(self.partitions[-1].get_id())
print("scale out one new partition with ID {}".format(self.partitions[-1].get_id()))
partition_key = self.partitions[-1].get_id()
self.partitions.sort(key=lambda x : x.get_id())
for i in range(len(self.partitions)):
if self.partitions[i].get_id() == partition_key:
new_partition = self.partitions[i]
if i + 1 < len(self.partitions):
next_partition = self.partitions[i+1]
else:
next_partition = self.partitions[0]
break
print("move elements from the next partition to the new partition")
for key in next_partition.get_items():
partition = self.get_partition(key)
if partition == new_partition:
print("move {} from partition {} to {}".format(key, next_partition.get_id(), new_partition.get_id()))
new_partition.put_item(key, next_partition.get_items()[key])
for key in new_partition.get_items():
next_partition.delete_item(key)
# remove one partition
def scale_in_partition(self):
if len(self.partitions) == 1:
print("Only one partition now. Cannot delete")
return
deleted_partition_index = randrange(len(self.partitions))
deleted_partition = self.partitions[deleted_partition_index]
if deleted_partition_index == len(self.partitions) - 1:
next_partition = self.partitions[0]
else:
next_partition = self.partitions[deleted_partition_index + 1]
print("Delete partition with ID {}. Move elements from the deleted partition to the next partition.".format(deleted_partition.get_id()))
for key,value in deleted_partition.get_items().items():
print("move {} from partition {} to {}".format(key, deleted_partition.get_id(), next_partition.get_id()))
next_partition.put_item(key, value)
self.available_hash_keys.append(deleted_partition.get_id())
del(self.partitions[deleted_partition_index])
def put_item(self, key, value):
partition = self.get_partition(key)
print("save {} to partition {}".format(key, partition.get_id()))
partition.put_item(key, value)
def get_item(self, key):
partition = self.get_partition(key)
print("get {} from partition {}".format(key, partition.get_id()))
return partition.get_item(key)
def delete_item(self, key):
partition = self.get_partition(key)
print("delete {} from partition {}".format(key, partition.get_id()))
partition.delete_item(key)
def get_item_count(self):
return functools.reduce(lambda x, y : x + y.get_item_count(), self.partitions, 0)
def get_hash_key(self, key):
return self.my_hash(key) % self.hash_space
def my_hash(self, text:str):
hash=0
for ch in text:
hash = ( hash*281 ^ ord(ch)*997) & 0xFFFFFFFF
return hash
def get_partition(self, key):
hash_key = self.get_hash_key(key)
if len(self.partitions) == 1 or hash_key <= self.partitions[0].get_id() or hash_key > self.partitions[-1].get_id() :
return self.partitions[0]
left = self.partitions[0].get_id()
for partition in self.partitions[1:]:
if hash_key <= partition.get_id() and hash_key > left:
return partition
left = partition.get_id()
return None
现在我们使用下面的代码对我们的DDB和Consistent_hash_manager进行测试：
from ddb import DDB
ddb = DDB()
table_name = "test_table"
ddb.create_table(table_name)
ddb.list_table()
ddb_table = ddb.get_table(table_name)
ddb_table.put_item("a", "value_of_a")
ddb_table.put_item("b", "value_of_b")
ddb_table.put_item("c", "value_of_c")
ddb_table.put_item("d", "value_of_d")
ddb_table.scale_out()
print(ddb_table.get_item("a"))
print(ddb_table.get_item("b"))
print(ddb_table.get_item("c"))
print(ddb_table.get_item("d"))
ddb_table.scale_in()
ddb_table.delete_item("a")
ddb_table.describe()我们首先创建了一个DDB表，表的默认partition只有一个。然后向其中插入了4个key/value，key分别是"a", "b", "c", "d"。然后我们给该表增加一个partition，此时我们应该会看到某些key/value被移动到新的partition里。然后我们查询key，观察它们是否是从正确的partition里查询的。然后我们删除掉一个partition，并且删除掉一个key，此时我们会发现有些数据会从被删除的partition移动到其它的partition里。最后我们查询一下目前表中key/value的数量。
我们测试结果如下：
# 创建表
Table name: test_table, item size: 0
# 添加key/value到partition 9中
save a to partition 9
save b to partition 9
save c to partition 9
save d to partition 9
# scale out一个partition 7
scale out one new partition with ID 7
# 三条数据被移动到新的partition里
move elements from the next partition to the new partition
move b from partition 9 to 7
move c from partition 9 to 7
move d from partition 9 to 7
# 读取key并且验证它们是从正确的partition中读取的
get a from partition 9
value_of_a
get b from partition 7
value_of_b
get c from partition 7
value_of_c
get d from partition 7
value_of_d
# scale in一个partition 9. partition 9里的数据被移动到partition 7里.
Delete partition with ID 9. Move elements from the deleted partition to the next partition.
move a from partition 9 to 7
# 删除一个key.并且查询目前表中key的数量.
delete a from partition 7
Table name: test_table, item size: 3
问题扩展
首先大家可以尝试在现在的实现基础上实现virtual node。我们在这里讨论另一个问题。如果我们有很多的partition，现在我们有一个key，我们如何快速的找到它对应的hash server（或者virtual node）呢？在我们目前的实现里我们是线性查找的，时间复杂度是O(n)。我们可以考虑使用二叉搜索树（binary search tree）来改善时间复杂度。二叉搜索树是一种二叉树，每一个node的左孩子节点的value都小于它，右孩子节点的value都大于它。这样我们搜索的效率就可以改善为O(lg(n))。
其次，我们还应该想到真正的产品实现会比我们的实现复杂的多。每一次scale out和scale in的操作都需要同步来保证数据的增删改读的可用性和正确性。
原文地址：https://blog.csdn.net/weixin_42325834/article/details/136134387
免责声明：本站文章内容转载自网络资源，如本站内容侵犯了原著者的合法权益，可联系本站删除。更多内容请关注自学内容网（zxcms.com）！