【總結系列】互聯網服務端技術體系：可擴展之數據分區

時間 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();
        }

    }

}