【总结系列】互联网服务端技术体系：可扩展之数据分区

时间 2020-11-25

标签 html java node 算法数据库数组缓存安全架构并发栏目 HTML 繁體版

原文原文链接

大而化小。html

引子

随着业务的快速发展，数据量也在飞速增加。单个存储节点的容量和并发读写都是有瓶颈的。怎么办呢？要解决这个问题，只要思考一个问题便可：在一亿个数中找一个数，和在一百个数中找一个数，哪一个更快？显然是后者。java

应对数据量膨胀的有效之法便是数据分区。将一个大的数据集划分为多个小的数据集分别进行管理，叫作数据分区。数据分区是分治策略的一种形式。本文总结关于数据分区的基本知识以及实践。node

总入口见：“互联网应用服务端的经常使用技术思想与机制纲要”

算法

基本知识

经常使用的数据分区方式有两种：范围分区和哈希分区。根据状况，还能够采用冷热分区、混合分区等。数据库

范围分区数组

将整个数据集按照顺序编号每 N 个一组划分红相对均匀的若干段连续范围。连续范围一般也是有序的。B+ 树的内节点和叶子节点实际上就是一个有序范围分区。范围分区能够有效支持范围查找。缓存

最经常使用的范围分区按时间分区。按时间分区有两个问题：安全

某个时间段的写负载都在一个分区上。为了不这一点，能够进一步按业务ID+时间戳来分区。
当数据量随着时间快速增加，一个时间分区的数据量也可能变得难以忍受。适用于数据规模增加不快的情形。
某段时间的读负载极可能也在一个分区上。

哈希分区架构

经过一个或多个哈希函数，对数据的 key 计算出哈希值，而后按照哈希值取模来落到某个分区。哈希分区能让数据分布更加均匀，但没法避免热点 key 的热点访问。哈希分区不支持范围查找。并发

采用哈希分区的例子： DB 按照业务 ID 取模进行分库分表； ES 按照业务 ID 取模进行分片。

冷热分区

将变化相对恒定的热数据单独放在一个分区里，将冷数据放在归档分区里。

冷热分区的例子：好比未完成订单相对于已完成订单是热数据，并且未完成订单的量在长期看来不会快速增加。所以，能够将未完成订单单独放在一个 ES 索引里（内部还能够分片），提供搜索。

混合分区

结合使用范围分区和哈希分区。可使用某个列的前缀来构建哈希分区，而使用整列数据来构建范围分区。固然，这也增大了存储空间开销和运维开销。

分区问题

分区字段

分区字段的选择一般遵循两个原则：

避免分区不均衡。对业务 ID 关联的记录数进行审查，若是某个业务 ID 可以关联的记录数可能占总记录数的比例很大，则按照该字段分区会存在分区数据不均衡问题。好比交易订单按店铺 ID 分，对于 VIP 大商家，就可能致使在某个分区上的热点数据；而按照用户分，则不会有，由于一个用户下单量是有限的，不会对总体产生影响。此外，若是对 2^m 取模分区，则 key 的低 m 位不能在短期内汇集性，好比都是 0001 - 0010。要作到分区均衡，一种方法是保持 key 的随机性。好比取 MD5 的一小段。
查询要求。梳理相应的查询请求，从中提取常见的查询字段。也能够经过分布式搜索引擎的方式来实现查询，使得分区字段选择不强依赖于查询请求。

分区数及大小

分区一般指的是逻辑分区，须要分配到物理节点上。一个物理节点一般有多个分区。要肯定分区数及分区大小。分区大小一般以某个数据量为最大限度。

要估算分区数，须要拿到一些基本数据：

预计要支撑多少读并发，写并发；要支撑多少年【规划值】；
健康的单分区/单节点所能支撑的数据行/记录数、读并发量、写并发量【经验值/监控值】；
当前总数据行/记录数、日增数据行/记录数；当前平均读并发量、平均写并发量、峰值读并发量、峰值写并发量【监控值】。能够监控每一年/月的数据行/记录数、读 QPS、写 QPS 的增加趋势状况，做为将来技术优化决策的依据。

热点数据

不管是范围分区仍是哈希分区，瞬时大并发的热点 key 的访问都是难以免和应对的。热点 key 访问的可考虑方案：

热点 key 的访问能够在热点 key 的基础上再加若干位，使得热点 key 的访问被打散，读的时候须要合并全部被打散到的分区；这样，分区的计算公式会相对复杂一点，并且不易扩展到其它 key 上。
经过实时计算自动检测到热点 key 的可能性，提早加载好缓存，或者作到自适应均衡负载。

辅助字段查询

辅助字段的查询，一般是先找到辅助字段所关联的分区字段（主键），再按分区字段进行查询。须要构建“辅助字段-分区字段”的映射信息。这个映射信息的存储和分区有两种方式：

关联到哪一个分区字段的值，就放在对应的分区里。辅助字段的某个值的映射信息可能分布在多个分区上。根据辅助字段查询时，须要合并全部分区的查询数据。好比说 name = qin 关联到两个 ID 1, 2；那么 name:1 放在 1 对应的分区里，name:2 放在 2 对应的分区里。辅助字段的分区与分区字段的分区是绑定的。 DB 采用这种作法。
单独为映射信息作统一的全局存储和分区。辅助字段的某个值的映射信息只在一个分区上。辅助字段的不一样值的映射可能在不一样的分区。好比说 name = qin 关联两个 ID 1,2 会做为一个总体放在某个分区里； name = ming 关联两个 ID 3,4 做为一个总体放在另外一个分区里。辅助字段的分区是单独设计的，与分区字段的分区无关。 ES 采用这种作法。

分区再均衡

当数据量/访问量剧增须要增长数据节点，或者机器宕机须要下线数据节点时，原有分区的数据须要在变动后的节点集合上从新分布。称为分区再均衡。

分区再均衡的方法 hash Mod N 。静态分区是采用固定分区数，动态分区则会增长或减小分区数。动态分区有利于让分区数据大小不超出某个最大限制。

分区再均衡有两种方案：

保持分区数不动，增长物理节点。使用 Steal Partition 的方法。DB 采用这种方法。ES 的主分片也是固定分片数。
保持物理节点不动，增大分区数。动态分区通常要应用哈希一致性算法。通常 K-V 存储用这种。

为何 DB 通常采用固定分区？由于 DB 每每要支持多个字段的查询，除了主字段分区之外，还要考虑辅字段分区。动态分区会增大这种复杂性。而 K-V 存储通常只支持主字段查询，没有额外要考虑。

分区应用

实际应用中，最多见的数据分区就是 DB 的分库分表。分库分表有水平和垂直两个维度。水平，一般是按行；垂直，一般是按业务或字段。水平分库，是将单个库的数据切分为多个库；水平分表，是将单个表的数据切分为多个表。库和表的 Schema 都是与原来彻底一致的。

那么，什么时候采用分库，什么时候采用分表呢？分库和分表的数量如何定？如何进行实际的分库分表操做？有哪些要注意的事项呢？

分库的缘由：单库的链接数和并发读写容量是有瓶颈的；此外多个业务争用同一个库的链接数，会相互影响；
分表的缘由：表的数据量太大， SQL 执行慢。
分库分表的基本标准判断：存储量 100G+ , 日增 20w+ , 单表数据量 1y+, 高峰期并发读写 1w+ 。

分库也能达到分表的效果。那么什么时候采用分表呢？若是表的数据量上涨，可是单库的并发读写容量并无多少上涨，则采用分表会更简单一些，运维成本应该也少一些。若是是由于须要支撑更多的并发读写，则首选分库，能足够解决并发读写的问题。单库的并发读写通常保持在 1000-2000 之间。分库以后，一般同时也实现了分表。若是不够，再细分表。分库分表的乘积数量一般选择 2 的幂次，由于在将数据分布到某个分区上时，须要进行取模操做，对 2 的 N 幂次取模只要取低 N 位便可。分库和分表也须要考虑好几年之用。通常 512, 1024 比较多。由于扩容时比较麻烦，须要进行分区再均衡。对于运行在线上的服务来讲，若是须要人工来作，风险会比较高。

分库分表的实际步骤：

STEP1: 开发。将原有的读写老库切换到读写新库。若是原来的访问 DB 层已经用 DAO 层隔离，那么改造的代码只要在 DAO 层切换库便可，上层的业务代码都不用动。所以，在作 DB 访问的时候，要注意 DAO 层的设计。
STEP2: 测试。测试包括两个部分： 1. 业务的全量回归； 2. 读写新库、读写老库。读新库和读老库的数据要作全字段数据对比，覆盖各类场景的数据。
STEP3: 部署。发布新的代码到线上。
STEP4: 数据迁移。以某个时间点为界，老库的全部数据必须迁移到新库中（最终统一读写新库），写入新库的数据要异步同步到老库里（回滚用）。数据迁移要考虑两个特殊的时间段： 1. 从老库切换到新库的一小段切换时间的新流量； 2. 回滚时重新库切回老库的一小段切换时间的新流量。

对于第一点来讲，要着重考虑数据不丢失、不重叠。要保证数据不丢失，则须要将切换的这一小段时间的数据积压在新库这边，待开启新库读写后，这段时间的流量直接进入新库，再同步到老库。切换的瞬间，中止老库的写。要保证数据不重叠，须要有惟一索引作保证，或者代码里作兼容，且重叠数据量很小。

对于第二点来讲，要考虑数据一致性。通常采用双写模式能够避免这一点。也就是，切换以后，异步写老库。这样，新流量老是进入老库。或者评估业务影响，若是短暂的不一致不影响业务的话，作到最终一致性亦可。

在分库分表以后，还须要分别考虑读写流量及相应的扩容。一般写主读从，读流量更多，保证扩容在从库上比较合适。由于从库不直接影响线上服务。

ES 的数据分区体如今分片（Shard）的概念。ES 的全部文档（Document）存储在索引（Index）里。索引是一个逻辑名字空间，包含一个或多个物理分片。一个分片的底层即为一个 Lucene 索引。ES 的全部文档均衡存储在集群中各个节点的分片中。ES 也会在节点之间自动移动分片，以保证集群的数据均衡。ES 分片分为主分片和复制分片。每一个文档都属于某个主分片。主分片在索引建立以后就固定了。复制分片用来实现冗余和容错，复制分片是可变的。

ES 对文档的操做是在分片的单位内进行的。实际上就是针对倒排索引的操做。倒排索引是不可变的，所以能够放在内核文件缓冲区里支持并发读。ES 更新文档必须重建索引，而不是直接更新现有索引。为了支持高效更新索引，倒排索引的搜索结构是一个 commit point 文件指明的待写入磁盘的 Segment 列表 + in-memory indexing buffer 。Segment 能够看作是 ES 的可搜索最小单位。新文档会先放在 in-memory indexing buffer 里。当文档更新时，新版本的文档会移动到 commit point 里，而老版本的文档会移动到 .del 文件里异步删除。ES 经过 fsync 操做将 Segment 写入磁盘进行持久化。因为 ES 能够直接打开处于文件缓冲区的 commit point 文件中的 Segment 进行查询（默认 1s 刷新），使得查询没必要写入磁盘后才能查询到，从而作到准实时。

ES 分片策略会影响 ES 集群的性能、安全和稳定性。ES 分片策略主要考虑的问题：分片算法如何？须要多少分片？分片大小如何？分片算法能够按照时间分区，也能够按照取模分区。分片数估算有一个经验法则：确保对于节点上已配置的每一个 GB，将分片数量保持在 20 如下。若是某个节点拥有 30GB 的堆内存，那其最多可有 600 个分片，可是在此限值范围内，设置的分片数量越少，效果就越好。通常而言，这能够帮助集群保持良好的运行状态。分片应当尽可能分布在不一样的节点上，避免资源争用。

HBase

HBase 的数据分区体如今 Region。 Region 是 HBase 均衡地存储数据的基本单元。Region 数据的均匀性，体如今 Rowkey 的设计上。 HBase Region 具备自动拆分能力，能够指定拆分策略，Region 在达到指定条件时会自动拆分红两个。能够指定的拆分策略有： IncreasingToUpperBoundRegionSplitPolicy 根据公式min(r^2*flushSize，maxFileSize) 肯定的大小；ConstantSizeRegionSplitPolicy Region 大小超过指定值 maxFileSize；DelimitedKeyPrefixRegionSplitPolicy 以指定分隔符的前缀 splitPoint，确保相同前缀的数据划分到同一 Region；KeyPrefixRegionSplitPolicy 指定 Rowkey 前缀来划分，适合于固定前缀。

除了 Region 自动拆分，还须要进行 Region 预分区。Region 预分区须要指定分为多少个 Region ，每一个 Region 的 startKey 和 endKey (底层会转化为 byte 数组)。若是数据可以比较均匀落到指定的 startKey 和 endKey, 就能够避免后续频繁的 Region Split。Region Split 虽然灵活，却会消耗集群资源，影响集群性能和稳定性。

HBase Region 的大小及数量的肯定，可参考业界实践 “HBase最佳实践之Region数量&大小”。官方推荐的 Regionserver上的 Region 个数范围在 20~200；每一个 Region 的大小在 10G~30G 之间。

分区示例

范围分区

假设有固定长度为 strlen 的字符串，字符取值集合限于 a-z ，且取值随机，要划分为 n 个分区。那么分区范围计算以下：

public class StringDividing {

    private static char[] chars = new char[] {
            'a','b','c','d','e','f','g','h','i','j','k','l','m','n','o','p','q','r','s','t','u','v','w','x','y','z'
    };

    private static final Integer CHAR_LEN = 26;

    public static List<String> divide(int strlen, int n) {
        int maxValue = maxValue(strlen);
        List<Integer> ranges = Dividing.divideBy(maxValue, n);
        return ranges.stream().map(num -> int2str(num, strlen)).collect(Collectors.toList());
    }

    public static int maxValue(int m) {
        int multiply = 1;
        while (m>0) {
            multiply *= CHAR_LEN;
            m--;
        }
        return multiply - 1;
    }

    /**
     * 将整型转换为对应的字符串
     */
    private static String int2str(int num, int strlen) {
        if (num < CHAR_LEN) {
            return nchars('a', strlen-1) + chars[num];
        }
        StringBuilder s = new StringBuilder();
        while ( num >= CHAR_LEN) {
            int r = num % CHAR_LEN;
            num = num / CHAR_LEN;
            s.append(chars[r]);
        }
        s.append(chars[num % CHAR_LEN]);
        return s.reverse().toString() + nchars('a', strlen-s.length());
    }

    private static String nchars(char c, int n) {
        StringBuilder s = new StringBuilder();
        while (n > 0) {
            s.append(c);
            n--;
        }
        return s.toString();
    }

    public static void main(String[] args) {
        for (int len=1; len < 6; len++) {
            divide(len,8).forEach(
                    e -> System.out.println(e)
            );
        }
    }
}

public class Dividing {

    public static List<Integer> divideBy(int totalSize, int num) {
        List<Integer> parts = new ArrayList<Integer>();
        if (totalSize <= 0) {
            return parts;
        }

        int i = 0;
        int persize = totalSize / num;
        while (num > 0) {
            parts.add(persize*i);
            i++;
            num--;
        }
        return parts;
    }
}

哈希分区

这里抽取了 Dubbo 的一致性哈希算法实现。核心是 TreeMap[Long, T] virtualNodes 的变动操做和 key 的哈希计算。

package zzz.study.algorithm.dividing;

import com.google.common.collect.Lists;

import java.io.UnsupportedEncodingException;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
import java.util.*;
import java.util.function.BiConsumer;
import java.util.function.Function;

public class ConsistentHashLoadBalance {

    public static void main(String[] args) {
        List<String> nodes = Lists.newArrayList("192.168.1.1", "192.168.2.25", "192.168.3.255", "255.255.1.1");

        ConsistentHashSelector<String> selector = new ConsistentHashSelector(nodes, Function.identity());
        test(selector);

        selector.addNode("8.8.8.8");
        test(selector);
    }


    private static void test(ConsistentHashSelector<String> selector) {
        Map<String, List<Integer>> map = new HashMap<>();
        for (int i=1; i < 16000; i+=1) {
            String node = selector.select(String.valueOf(i));
            List<Integer> objs = map.getOrDefault(node, new ArrayList<>());
            objs.add(i);
            map.put(node, objs);
        }
        map.forEach(
                (key, values) -> {
                    System.out.println(key + " contains: " + values.size() + " --- " + values);
                }
        );
    }

    private static final class ConsistentHashSelector<T> {

        private final TreeMap<Long, T> virtualNodes;
        private final int replicaNumber = 160;
        private final Function<T, String> keyFunc;

        ConsistentHashSelector(List<T> nodes, Function<T, String> keyFunc) {
            this.virtualNodes = new TreeMap<Long, T>();
            this.keyFunc = keyFunc;
            assert keyFunc != null;
            for (T node : nodes) {
                addNode(node);
            }
        }

        public boolean addNode(T node) {
            opNode(node, (m, no) -> virtualNodes.put(m,no));
            return true;
        }

        public boolean removeNode(T node) {
            opNode(node, (m, no) -> virtualNodes.remove(m));
            return true;
        }

        public void opNode(T node, BiConsumer<Long, T> hashFunc) {
            String key = keyFunc.apply(node);
            for (int i = 0; i < replicaNumber / 4; i++) {
                byte[] digest = md5(key + i);
                for (int h = 0; h < 4; h++) {
                    long m = hash(digest, h);
                    hashFunc.accept(m, node);
                }
            }
        }

        public T select(String key) {
            byte[] digest = md5(key);
            return selectForKey(hash(digest, 0));
        }

        private T selectForKey(long hash) {
            Map.Entry<Long, T> entry = virtualNodes.ceilingEntry(hash);
            if (entry == null) {
                entry = virtualNodes.firstEntry();
            }
            return entry.getValue();
        }

        private long hash(byte[] digest, int number) {
            return (((long) (digest[3 + number * 4] & 0xFF) << 24)
                    | ((long) (digest[2 + number * 4] & 0xFF) << 16)
                    | ((long) (digest[1 + number * 4] & 0xFF) << 8)
                    | (digest[number * 4] & 0xFF))
                    & 0xFFFFFFFFL;
        }

        private byte[] md5(String value) {
            MessageDigest md5;
            try {
                md5 = MessageDigest.getInstance("MD5");
            } catch (NoSuchAlgorithmException e) {
                throw new IllegalStateException(e.getMessage(), e);
            }
            md5.reset();
            byte[] bytes;
            try {
                bytes = value.getBytes("UTF-8");
            } catch (UnsupportedEncodingException e) {
                throw new IllegalStateException(e.getMessage(), e);
            }
            md5.update(bytes);
            return md5.digest();
        }

    }

}

小结

分治是最为基本的计算机思想之一。而数据分区是应对海量数据处理的基本前提。常见数据分区有范围分区和哈希分区两种，根据状况选用。

分区是逻辑概念。分区每每会分布到多个机器节点上。数据分区要考虑数据均匀分布问题、分区大小及分区数。数据分区加冗余，构成了高可用分布式系统的基础。

【总结系列】互联网服务端技术体系：可扩展之数据分区

引子

基本知识

分区问题

分区应用

分区示例

小结

参考资料