深入理解分布式系统的 7 种数据分片策略

数据分片是指将全量的数据通过某种计算规则分别存放到多个数据存储上，以平摊单个系统的存储压力和读写压力，实现数据存储上的线性扩展能力，实现系统读写性能的线性扩展能力。

常见的数据中间件中都使用一种或几种数据分片方式，例如MySQL、Redis、HBase、ElasticSearch、Kafka、Hive等都会进行数据分片，数据分片共有7种方式，分别为 Hash取余分片、一致性Hash分片、Range分片、时间分片、固定行数分片、固定文件大小分片、随机分片。

Range 范围分片

Range 范围分片是使用索引表维护每个节点负责的起始范围，数据读写时查询索引表，将请求路由到对应的存储节点。

HBase 范围分片是怎么回事？

为了更好理解，我以HBase为例，HBase每张表在存储时会切分成多个Region，Region分别存储在多个节点上。其中分片key是rowKey，即每条记录主键。rowKey要求为ascii码，业务一般设置为英文和数字，如何将rowKey路由到对应的Region呢？每个Region都有自己的负责的startKey、endKey。可以在建表时指定每个region的范围，也可以指定排序方式。

例如创建50个分区，并且指定rowKey排序方式为16进制字符串方式分割。HBase会生成对应的索引表。

create'test',{NAME => 'f1',COMPRESSION => 'snappy' }, { NUMREGIONS => 5, SPLITALGO => 'HexStringSplit' }

• HexStringSplit：适用于以十六进制的字符串作为前缀的Rowkey。
• DecimalStringSplit：适用于以十进制的数字字符串作为前缀的Rowkey。
• UniformSplit：适用于Rowkey的前缀是完全随机的。

也可以指定每个分区的范围，创建6个分区，每个Region负责以下范围。

create 'datamanroad:Employee', 'info', 'partition1', SPLITS => ['10000','20000','30000','40000','50000']

例如第一个分区负责[, 10000] 起始到10000，最后一个负责[50000,] 50000到结束。每个 Region 负责一个范围。

HBase在数据读写时，便根据索引表将请求映射到对应的Region。并且HBase默认会自动进行Region管理，当Region数量过大时，会自动分裂，分裂后每个Region各负责一部分RowKey范围。每个Region内部rowKey也是有序存储的，不同于MySQL使用B+树索引，HBase使用LSM（Log-Structured Merge Tree日志结构合并树）树，适合于频繁插入和更新的索引场景。

为什么 Hbase 使用范围分片

为什么HBase要支持Region范围分片，本质还是因为HBase除随机查找外，还支持范围查找。这就要求存储上必须是有序的，不光每个Region内部有序，每个Region之间也要能排序，否则一次范围查找的场景，HBase就要查找所有Region，这将大大降低查询效率。

所以不同的使用场景，决定了使用不同的分片策略。

RedisCluster如果使用范围分片呢？

RedisCluster 如果使用范围分片会有哪些好处呢？如果可以指定路由表，即每个Redis节点负责某一个范围，我们就可以将同一批Key路由到同一个Redis节点。

因为RedisCluster 将key hash到不同的节点，所以像Lua和mget等涉及多个Key的请求会失败。但假设RedisCluter支持范围分片，例如Redis 节点 1 负责A - G范围，那么 A0001_type1， A0001_type2 两个Key就可以路由到同一个Redis节点，那么LUA脚本和mget等涉及多个Key 操作就可以在RedisCluster中使用。

例如文章的总阅读量、每天阅读量数据保存在Redis中，使用incrBy实时更新。两个key如果在同一个节点，就可以使用lua脚本同时对这两个Key增加，无需调用两遍。同时查询时也可以使用mget批量查询多个key。然而现实是redis-cluster使用hash分片，上述操作都无法一次执行，只能逐个操作。

虽然RedisCluster不支持范围分片，但是Redis 客户端可以自行分片。ShardedJedis支持keyTagPattern模式，即抽取key的一部分keyTag做sharding，通过命名key的格式，将一组相关联的key放入同一个Redis节点，从而实现批量操作。

不同的使用场景决定了使用不同的分片策略。

Hash 取模适合哪些场景？

Hash是最常见的分片方式，通过Hash算法把Key均匀打散生成一个数值，

例如订单数据的数据库分库分表大多数使用userId 取模，路由到对应的库和表。

由于userId的生成基本上是均匀的，后两位随机排列，所以直接使用userId取模即可得到分片值。但假设要hash的是key为不均匀的数字呢？ 例如 后两位为偶数的概率非常大，那么按照100取模，数据分布就会非常倾斜。

MurmurHash是什么？

目前最常用的hash算法是 murmurhash。

murmurhash 广泛应用于各开源产品Java 界中 Redis，Memcached，Cassandra，Hadoop，HBase，Lucene，spark，nginx等。

我做了一个对比测试，对比 Long.hashCode和 MurmurHash对偶数进行hash的效果。

生成一组偶数数字，从0-100W，共50W个数字，分别使用java Long.hashCode和 MurMurHash进行hash，将hash值对10取模，最终得到神奇的测试结果。

可以看到 使用MurmurHash对50W的偶数进行散列，分别散列到10个桶里，基本上是均匀的，每个桶在5W上下，差异数量不超过200。然而 Java Long.HashCode生成的散列值就奇特了。

使用Java Long.hashCode，hash值和原值一样。可以看到我希望将偶数散列到10个桶里面，因为Long hash值和原值一样，所以10个桶只有5个偶数桶有数据。

究其原因是因为Java Long.HashCode是对Long类型高 32 位和低 32 位 按位进行异或。因为我选取的值都小于 2的32次方，所以高32位都是0。相当于一个数和 0异或，结果自然是自己（按位异或：相同则为0，不同则为1）。得到的hash值自然也都是偶数。被分到10个桶，数据肯定非常倾斜。

由此可见，如果需要对数据进行hash分片，一定要确认hash值是随机分散的，如果不分散极可能导致数据倾斜非常严重。最好亲自测试一下使用的hash方法，避免得到倾斜严重的 hash 分布。

以下为本次测试代码

    com.google.guava
    guava
    31.1-jre

long max = 1000000;
Map map = Maps.newHashMap();
int mo = 10;
for (long i = 0; i < mo; i++) {
   map.put(i, 0L);
}

Map map2 = Maps.newHashMap();
for (long i = 0; i < mo; i++) {
   map2.put(i, 0L);
}

for (long i1 = 0; i1 < max; i1 += 2) {
   HashFunction hashFunction = Hashing.murmur3_128();
   HashCode code = hashFunction.hashLong(i1);
   long v = code.asLong();
   if (v < 0) {
      v = 0 - v;
   }
   Long count = map.get(v % mo);
   count++;
   map.put(v % mo, count);

   long v2 = Long.valueOf(i1).hashCode();
   if (v2 < 0) {
      v2 = 0 - v2;
   }
   count = map2.get(v2 % 10);
   count++;
   map2.put(v2 % 10, count);
}
System.out.println("murmurhash:" + map);
System.out.println("Long hashCode:" + map2);

如果要分片的key为字符串呢？字符串是无法取模的，此时必须使用hash算法对key求hash值了，目的是得到更加均匀、随机的hash值。

我使用了String.hashCode 和 MurmurHash进行测试，发现两者hash效果差不多。

生成100W的随机字符串，分别以上两种方式来计算hash值。效果如下，差异并不大，散列随机效果非常好。一共10个桶，每个桶分到10W上下，差异值不超过300。

总结：如果key本身是散列均匀的数字，无需hash直接取模即可，否则最好使用MurmurHash hash后再取模，这样数据会比较均匀。

一致性Hash

Hash取模有什么痛点？

hash取模的方式简单实用，适用于存储节点数量长期不变的场景。在遇到数据扩缩容场景问题，hash取模算法表现很差。存储负载过大或过小需要扩缩容，例如将10个节点扩容为15个，按照之前的取模算法，hashValue%15，将有大量数据的分片和之前不同，甚至几乎全量的数据需要迁移到新节点，系统扩缩容时数据迁移的难度是巨大的，迁移过程稳定性的挑战也是巨大的。那么扩缩容时有什么办法能减少数据迁移量呢？

什么是一致性Hash？

一致性Hash就是为了解决hash取模算法的痛点提出的，可有效减少扩缩容场景数据漂移情况。一致性hash提出虚拟节点的概念，即增大hash取模的数值范围，将取模后的值作为虚拟节点值，然后提供映射表将虚拟节点值再映射为实际节点值。当需要新增和减少实际节点时，只需要修改一部分虚拟节点的映射表即可。

例如当前划分1W的虚拟节点，共4台机器，即4个实际节点。当前系统的映射表如下

此时新增了节点5，需要修改一部分虚拟节点的映射。

可以从以上 4 个节点中拼凑2000个节点划拨给 节点5。这么一看5个节点各自负责2000个虚拟节点，实现了数据均衡分布。有人会好奇，怎么实现这个映射表呢? 很容易，由于虚拟节点的数量是可以确定的，可维护1W个KV的HashMap，每次变更实际节点数时，需要有调度模块，重新分配每个实际节点要负责的虚拟节点。生成对应的迁移计划，然后迁移完成后修改映射表即可。

这个思想就是 一致性Hash。网上很多一致性Hash的资料 提到了Hash环。

Hash环思想是将所有的虚拟节点映射到整个环上，每个实际节点负责和上一个节点之间的所有节点。每次节点加入只需要分摊一部分虚拟节点即可，无需重新计算整个Hash环。

但在我看来，这个Hash环的解释不容易理解，也不切合实际。由于变更节点数量时，需要修改虚拟到实际节点的映射表，为了尽可能的减少迁移量，又要兼顾尽可能均分所有的虚拟节点，势必要从每个实际节点中迁移一部分虚拟节点，这就导致每个实际节点负责的虚拟节点值并不总是连续的。 所以hash环的思路并不是很切合实际。

RedisCluster中的一致性Hash

RedisCluster使用了一致性hash的思想。RedisCluster 将多个Redis节点组合为一个集群，需要将缓存Key分片到每个节点。 缓存Key使用CRC16进行hash，然后被分片到2^14=16384个槽，CRC16(key) & 16384。这16384个槽即是16384个虚拟节点，运维人员可以指定每个Redis节点负责的槽范围，也可以交由redis-trib.rb运维工具进行管理。

使用redis-cli管理每个节点的槽范围

redis-cli -h 192.168.0.1 –p 6379 cluster addslots 0,4095 
redis-cli -h 192.168.0.2 –p 6379 cluster addslots 4096,8191

也可以 使用redis-trib.rb运维工具创建集群，运维工具会自动帮你均匀的分配好每个节点的虚拟槽数量

redis-trib.rb create --replicas 1 127.0.0.1:6379 127.0.0.1:6380 127.0.0.1:6381 127.0.0.1:6382 127.0.0.1:6383 127.0.0.1:6384

16384个槽分配成功，集群创建完成。

使用redis-trib.rb运维工具还再平衡 槽的数量，可以指定每个节点的权重，分别分配不同的槽数量。平衡过程中，完成槽的数据迁移。

rebalance       host:port
                  --weight 
                  --auto-weights
                  --use-empty-masters
                  --timeout 
                  --simulate
                  --pipeline 
                  --threshold 
                  
# redis-trib.rb rebalance --weight a8b3d0f9b12d63dab3b7337d602245d96dd55844=3 --weight f413fb7e6460308b17cdb71442798e1341b56cbc=2  --use-empty-masters  127.0.0.1:6379

当新增节点时，redis-trib.rb 还可以指定给目标节点迁移制定数量的虚拟槽。例如分配100个槽，这100个虚拟槽就是从其他节点中拼凑来的，而不是连续划分的，目的是保证每个节点的槽数量尽可能均匀。

一致性 Hash 还有哪些应用？

一致性Hash在很多地方都使用着，例如Dubbo就提供了一致性Hash的负载均衡策略，尽可能保证请求被路由到固定的节点。如果服务节点有大数据量的本地缓存，但是每个节点不足以保存全量的本地缓存。如果使用一致性Hash，每部分机器负责一部分用户的缓存，将这部分用户请求路由到这些节点上，保证被路由的节点最大概率有对应用户的本地缓存。这样可以大大提高缓存命中率，同时减少机器上本地缓存的内存压力。

总结：一致性Hash将Key映射为虚拟节点，同时维护虚拟节点到实际节点的映射表。每次扩缩容节点，只需要修改映射表即可，迁移一部分数据给新节点。避免全量数据的漂移。

以上Hash取模、一致性Hash、Range范围三种分片方式是最常用的分片策略，除此之外还存在4个分片策略。

基于时间分片

如果数据的时间属性非常重要，而且查询时基本上会指定时间范围，就可以使用时间分片。例如指定查询具体时间范围的日志，在查找时就可以查指定日期的分片数据。

Hive如何基于时间分片

最典型的当属Hive，hive在创建表时可以指定时间作为分区键，一般为日期。把数据按照每一天组织起来，在SQL查询时 指定时间范围，这样可以避免查询全量的数据。

CREATE TABLE IF NOT EXISTS `$target.table`(
    user_id                    bigint COMMENT '用户ID'
)COMMENT '备注' 
PARTITIONED BY (dt string COMMENT 'ctime, 日期分区字段，格式为datekey(yyyymmdd)')
STORED AS ORC
;

INSERT OVERWRITE TABLE `${target.table}` PARTITION (dt) SELECT * FROM XXX;

以上Hive建表SQL指定了日期作为分区键，要求SQL必须指定日期范围查询数据。

日志可以按时间分片吗？

除此之外还有很多时间分片的场景，例如应用日志，一般通过小时、天来切分日志，每个小时的日志分到不同的文件中，避免生成过大的日志文件，从而增加上传和清理的成本。

业务系统有哪些场景可以使用时间分片？

我们业务方案设计中也会涉及到时间分片的应用，例如需要存储用户浏览记录时，可以按天存储浏览记录和每天的浏览数，分页查找时就能避免查询全量的数据。具体参考# 10W+TPS高并发场景【我的浏览记录】系统设计

ElasticSearch 时间分片场景

一般使用 ElasticSearch 实现日志检索，在查询时必须指定时间范围，为什么呢？因为是为了缩小要检索的数据范围，使用ElasticSearch实现数据检索时并不会存储全量的数据，而是存储近期的数据，例如三个月，半年。过期的数据会被清理。此时就可以使用时间分片，每天的数据储存在一个新索引上，过期会清理。用户在检索时也必须指定时间范围，以便指定具体日期的索引进行检索。通过日期组织索引数据是常见的方案。

随机分片

随机分片虽不起眼，但是在 Kafka 中却被使用着。随机分片就是不控制数据被路由到哪个分片，随缘。典型的例子是kafka的分片策略，Kafka每个 topic 下划分为多个 partition，partition 可以动态添加，每个消息会被投递一个 partition，消费时每个partition 只能被一个消费者实例消费。Kafka 的分片路由策略比较复杂，默认情况下Kafka会随机路由消息到一个 partition。 Kafka 还提供了轮训策略，即把消息平均路由到每一个分片。

此外Kafka 也提供了指定分区的策略，即由生产者自定义路由策略，选择自己要发送的分区。这时候用户可以选择使用 Hash 路由方式，例如按照用户 UserId，把同一个用户的消息路由到同一个分片，由于只有一个消费者实例消费这个分片，所以可保证单个用户的消息可以顺序串行消费。

Kafka 为什么可以选择随机分片？

Kafka 和 MySQL、Redis、HBase 等中间件均不同，Kafka 不对外提供随机查找、范围查找的能力。只需要保证消息可靠的被投递到消费者，消费者按序逐个消费。所以 Kafka 没有随机查找的诉求，自然不需要Hash 取模和一致性 Hash、范围查找等分片策略。Kafka 生产端会等待一个窗口期，批量把窗口期的消息随机路由到一个分片，如果默认按照 Hash 分片，一个窗口期内的消息被路由到不同分片，就无法保证批量消息投递、同时落盘存储的可靠性，也降低了消息发送的吞吐量。

当然 Kafka 提供了灵活性，业务可自定义路由分片策略，满足业务层特殊的逻辑需要。

不同的场景诉求决定了使用不同的路由策略。

按文件大小分片

业务系统中一般不涉及按照文件大小分片，但是在存储系统中涉及到读写文件，为了避免文件过大降低读写文件时性能，会控制文件的大小。例如 Kafka 每个分片对应一个文件目录，每个分片目录下都包含多个文件，每个文件包含了一部分消息。当文件数量超过阈值时，Kafka 就会重新新建一个文件，消息也会写到新的文件中。

.log文件保存了消息内容，而 index 文件是消息的索引文件，两者除后缀外同名，index 文件用来标识消息 offset 和对应文件内偏移量。当然index 文件采取稀疏索引存储方式，它减少索引文件大小，只记录了一部分消息的位置。

例如 8,1686 代表文件内第八条消息，在文件物理偏移 1686 位开始。这样 Kafka 在检索一条消息时，就能兼顾性能和存储。

Kafka 为什么要切分日志文件

Kafka 虽然不提供消息 Id 的随机查询接口，但是提供重置消费位点的能力，消费组可以指定某个分区的消费点及 offset，kafka 需要尽快的定位到该条消息。

试想一下，如果日志文件巨大，文件读取时定位到该消息的 offset就需要很久。如果切分为一个个小文件，再辅助文件索引，就可以最快速的定位到该条消息，然后从该条消息开始消费。

实际上除了 Kafka 会切分文件，HBase 也会切分文件。前面提到 HBase 每个 Region 是一个分区，可以预分区，提前规划好分区。HBase 也提供了默认分区策略，即按照Region 大小，自动分裂。

HBase Region 按文件大小切分。

HBase 每个 Region 超过一个阈值，会自动分裂。该阈值的计算比较复杂。
Math.min( regionNumber ^ 3 * hbase.hregion.memstore.flush.size *2, 默认最大文件大小  )

使用默认值替换后，为
Math.min( regionNumber ^ 3 * 256 M, 10G)

例如 hbase.hregion.memstore.flush.size = 128M。

为什么 Region 的阈值定的这么复杂，主要是为了保证当数据量较少时，切分的阈值较少、数据量大的场景阈值较大。例如256M-2G期间，只会有两个 Region，而数据量越来越大时，阈值就要越大，直到10G。避免小数据量场景，却出现几十个 Region 过度分片的情况发生。

不同的场景决定了不同的数据分片策略。

按照文件行数分片

除了按照文件大小切分，对于一些按行分割的文件，可以选择按照固定行数切分文件，例如 1亿条数据的大文件，切分为 10 个 1000W 条数据的小文件，然后分别读取该文件，并行处理。

这种分片策略本质上和按照文件大小进行切分类似。都是避免生成过大的文件。一般情况下 可以结合两者，例如首先按照文件大小进行切分，同时限定最大行数，超过最大文件大小、最大文件行数任一条件，就进行文件切分。

例如在本文中，# 10亿条记录插入数据库用多久？ 10 亿数据的文件就进行了按行数进行了切分，创建 100 个读写任务并行处理 100 个小文件，大大提高数据处理速度。

10 亿的大文件受限于一块磁盘的读写性能，只能顺序读取。如果把文件切分为 100 份，就可以增大数据读取的并发度，提高读取性能。

所以在文件生产阶段就应该按照文件行数、或者文件大小进行切分。而不应该等到读取文件前，在进行文件切分。

从上面7 种数据分片方式我们能总结什么呢？

不同的场景决定了不同的数据分片策略。

总结