詳解大數據框架之分區，分桶，分片
本文轉載自微信公眾號「大數據左右手」，作者王了個博 。轉載本文請聯系大數據左右手公眾號。
前言
在大數據分布式中，分區，分桶，分片是設計框架的重點。此篇就來總結各個框架。
目錄
- Hive分區與分桶
- ES分片
- Kafka分區
- HBase分區
- Kudu分區
Hive
Hive分區
是按照數據表的某列或者某些列分為多區，在hive存儲上是hdfs文件，也就是文件夾形式。現在最常用的跑T+1數據，按當天時間分區的較多。
把每天通過sqoop或者datax拉取的一天的數據存儲一個區，也就是所謂的文件夾與文件。在查詢時只要指定分區字段的值就可以直接從該分區查找即可。創建分區表的時候，要通過關鍵字 partitioned by (column name string)聲明該表是分區表，并且是按照字段column name進行分區，column name值一致的所有記錄存放在一個分區中，分區屬性name的類型是string類型。
當然，可以依據多個列進行分區，即對某個分區的數據按照某些列繼續分區。
向分區表導入數據的時候，要通過關鍵字partition((column name="xxxx")顯示聲明數據要導入到表的哪個分區
設置分區的影響
首先是hive本身對分區數有限制，不過可以修改限制的數量;
- set hive.exec.dynamic.partition=true;
- set hive.exec.max.dynamic.partitions=1000;
- set hive.exec.dynamic.partition.mode=nonstrict;
- set hive.exec.parallel.thread.number=264;
hdfs對單個目錄下的目錄數量或者文件數量也是有限制的，也是可以修改的;
NN的內存肯定會限制，這是最重要的，如果分區數很大，會影響NN服務，進而影響一系列依賴于NN的服務。所以最好合理設置分區規則，對小文件也可以定期合并，減少NN的壓力。
Hive分桶
在分區數量過于龐大以至于可能導致文件系統崩潰時，我們就需要使用分桶來解決問題
分桶是相對分區進行更細粒度的劃分。分桶則是指定分桶表的某一列，讓該列數據按照哈希取模的方式隨機、均勻的分發到各個桶文件中。因為分桶操作需要根據某一列具體數據來進行哈希取模操作，故指定的分桶列必須基于表中的某一列(字段) 要使用關鍵字clustered by 指定分區依據的列名，還要指定分為多少桶
create table test(id int,name string) cluster by (id) into 5 buckets .......
insert into buck select id ,name from p cluster by (id)
Hive分區分桶區別
- 分區是表的部分列的集合，可以為頻繁使用的數據建立分區，這樣查找分區中的數據時就不需要掃描全表，這對于提高查找效率很有幫助
- 不同于分區對列直接進行拆分，桶往往使用列的哈希值對數據打散，并分發到各個不同的桶中從而完成數據的分桶過程
- 分區和分桶最大的區別就是分桶隨機分割數據庫，分區是非隨機分割數據庫
ElasticSearch分片
主分片：用于解決數據水平擴展的問題，一個索引的所有數據是分布在所有主分片之上的(每個主分片承擔一部分數據，主分片又分布在不同的節點上)，一個索引的主分片數量只能在創建時指定，后期無法修改，除非對數據進行重新構建索引(reindex操作)。
副本分片：用于解決數據高可用的問題，一個副本分片即一個主分片的拷貝，其數量可以動態調整，通過增加副本分片也可以實現提升系統讀性能的作用。
在集群中唯一一個空節點上創建一個叫做 blogs 的索引。默認情況下，一個索引被分配 5 個主分片
- {
- "settings": {
- "number_of_shards": 5,
- "number_of_replicas": 1
- }
- }
到底分配到那個shard上呢?
- shard = hash(routing) % number_of_primary_shards
routing 是一個可變值，默認是文檔的 _id ，也可以設置成一個自定義的值。routing 通過 hash 函數生成一個數字，然后這個數字再除以 number_of_primary_shards (主分片的數量)后得到余數 。這個在 0 到 number_of_primary_shards 之間的余數，就是所尋求的文檔所在分片的位置。
如果數量變化了，那么所有之前路由的值都會無效，文檔也再也找不到了
- 分片過少
如15個節點，5個主分片，1個副本 會造成每個索引最多只能使用10個節點(5個主分片，5個從分片)，剩余5節點并沒有利用上;資源浪費
如：3節點;3分主分片，1副本 當數據量較大的時，每個分片就會比較大
- 分片過多
- 創建分片慢：es創建分片的速度會隨著集群內分片數的增加而變慢。
- 集群易崩潰：在觸發es 自動創建Index時，由于創建速度太慢，容易導致大量寫入請求堆積在內存，從而壓垮集群。
- 寫入拒絕：分片過多的場景中，如果不能及時掌控業務的變化，可能經常遇到單分片記錄超限、寫入拒絕等問題。
分片的注意事項
避免使用非常大的分片，因為這會對群集從故障中恢復的能力產生負面影響。對分片的大小沒有固定的限制，但是通常情況下很多場景限制在 30GB 的分片大小以內。
當在ElasticSearch集群中配置好你的索引后, 你要明白在集群運行中你無法調整分片設置. 既便以后你發現需要調整分片數量, 你也只能新建創建并對數據進行重新索引.
如果擔心數據的快速增長, 建議根據這條限制: ElasticSearch推薦的最大JVM堆空間 是 30~32G, 所以把分片最大容量限制為 30GB, 然后再對分片數量做合理估算。例如, 如果的數據能達到 200GB, 則最多分配7到8個分片。
kafka分區
生產者
分區的原因
- 方便在集群中擴展，每個Partition可以通過調整以適應它所在的機器，而一個topic又可以有多個Partition組成，因此整個集群就可以適應任意大小的數據了;
- 可以提高并發，因為可以以Partition為單位讀寫了。
分區的原則
- 指明 partition 的情況下，直接將指明的值直接作為 partiton 值;
- 沒有指明 partition 值但有 key 的情況下，將 key 的 hash 值與 topic 的 partition 數進行取余得到 partition 值;
- 既沒有 partition 值又沒有 key 值的情況下，第一次調用時隨機生成一個整數(后面每次調用在這個整數上自增)，將這個值與 topic 可用的 partition 總數取余得到 partition 值，也就是常說的 round-robin 算法。
消費者
分區分配策略
一個consumer group中有多個consumer，一個 topic有多個partition，所以必然會涉及到partition的分配問題，即確定那個partition由哪個consumer來消費 Kafka有三種分配策略，一是RoundRobin，一是Range。高版本還有一個StickyAssignor策略 將分區的所有權從一個消費者移到另一個消費者稱為重新平衡(rebalance)。當以下事件發生時，Kafka 將會進行一次分區分配：
- 同一個 Consumer Group 內新增消費者
- 消費者離開當前所屬的Consumer Group，包括shuts down 或 crashes
Range分區分配策略
Range是對每個Topic而言的(即一個Topic一個Topic分)，首先對同一個Topic里面的分區按照序號進行排序，并對消費者按照字母順序進行排序。然后用Partitions分區的個數除以消費者線程的總數來決定每個消費者線程消費幾個分區。如果除不盡，那么前面幾個消費者線程將會多消費一個分區。假設n=分區數/消費者數量，m=分區數%消費者數量，那么前m個消費者每個分配n+1個分區，后面的(消費者數量-m)個消費者每個分配n個分區。假如有10個分區，3個消費者線程，把分區按照序號排列
0，1，2，3，4，5，6，7，8，9
消費者線程為
C1-0，C2-0，C2-1
那么用partition數除以消費者線程的總數來決定每個消費者線程消費幾個partition，如果除不盡，前面幾個消費者將會多消費一個分區。在我們的例子里面，我們有10個分區，3個消費者線程，10/3 = 3，而且除除不盡，那么消費者線程C1-0將會多消費一個分區，所以最后分區分配的結果看起來是這樣的：
C1-0：0，1，2，3
C2-0：4，5，6
C2-1：7，8，9
如果有11個分區將會是：
C1-0：0，1，2，3
C2-0：4，5，6，7
C2-1：8，9，10
假如我們有兩個主題T1,T2，分別有10個分區，最后的分配結果將會是這樣：
C1-0：T1(0，1，2，3) T2(0，1，2，3)
C2-0：T1(4，5，6) T2(4，5，6)
C2-1：T1(7，8，9) T2(7，8，9)
RoundRobinAssignor分區分配策略
RoundRobinAssignor策略的原理是將消費組內所有消費者以及消費者所訂閱的所有topic的partition按照字典序排序，然后通過輪詢方式逐個將分區以此分配給每個消費者. 使用RoundRobin策略有兩個前提條件必須滿足：
同一個消費者組里面的所有消費者的num.streams(消費者消費線程數)必須相等;
每個消費者訂閱的主題必須相同。
加入按照 hashCode 排序完的topic-partitions組依次為
T1-5, T1-3, T1-0, T1-8, T1-2, T1-1, T1-4, T1-7, T1-6, T1-9
我們的消費者線程排序為
C1-0, C1-1, C2-0, C2-1
最后分區分配的結果為：
C1-0 將消費 T1-5, T1-2, T1-6 分區
C1-1 將消費 T1-3, T1-1, T1-9 分區
C2-0 將消費 T1-0, T1-4 分區
C2-1 將消費 T1-8, T1-7 分區
StickyAssignor分區分配策略
Kafka從0.11.x版本開始引入這種分配策略，它主要有兩個目的：
- 分區的分配要盡可能的均勻，分配給消費者者的主題分區數最多相差一個
- 分區的分配盡可能的與上次分配的保持相同。
當兩者發生沖突時，第一個目標優先于第二個目標。鑒于這兩個目的，StickyAssignor策略的具體實現要比RangeAssignor和RoundRobinAssignor這兩種分配策略要復雜很多。
假設消費組內有3個消費者
C0、C1、C2
它們都訂閱了4個主題：
t0、t1、t2、t3
并且每個主題有2個分區，也就是說整個消費組訂閱了
t0p0、t0p1、t1p0、t1p1、t2p0、t2p1、t3p0、t3p1這8個分區
最終的分配結果如下：
消費者C0：t0p0、t1p1、t3p0
消費者C1：t0p1、t2p0、t3p1
消費者C2：t1p0、t2p1
這樣初看上去似乎與采用RoundRobinAssignor策略所分配的結果相同
此時假設消費者C1脫離了消費組，那么消費組就會執行再平衡操作，進而消費分區會重新分配。如果采用RoundRobinAssignor策略，那么此時的分配結果如下：
消費者C0：t0p0、t1p0、t2p0、t3p0
消費者C2：t0p1、t1p1、t2p1、t3p1
如分配結果所示，RoundRobinAssignor策略會按照消費者C0和C2進行重新輪詢分配。而如果此時使用的是StickyAssignor策略，那么分配結果為：
消費者C0：t0p0、t1p1、t3p0、t2p0
消費者C2：t1p0、t2p1、t0p1、t3p1
可以看到分配結果中保留了上一次分配中對于消費者C0和C2的所有分配結果，并將原來消費者C1的“負擔”分配給了剩余的兩個消費者C0和C2，最終C0和C2的分配還保持了均衡。
如果發生分區重分配，那么對于同一個分區而言有可能之前的消費者和新指派的消費者不是同一個，對于之前消費者進行到一半的處理還要在新指派的消費者中再次復現一遍，這顯然很浪費系統資源。StickyAssignor策略如同其名稱中的“sticky”一樣，讓分配策略具備一定的“粘性”，盡可能地讓前后兩次分配相同，進而減少系統資源的損耗以及其它異常情況的發生。
到目前為止所分析的都是消費者的訂閱信息都是相同的情況，我們來看一下訂閱信息不同的情況下的處理。
舉例，同樣消費組內有3個消費者：
C0、C1、C2
集群中有3個主題：
t0、t1、t2
這3個主題分別有
1、2、3個分區
也就是說集群中有
t0p0、t1p0、t1p1、t2p0、t2p1、t2p2這6個分區
消費者C0訂閱了主題t0
消費者C1訂閱了主題t0和t1
消費者C2訂閱了主題t0、t1和t2
如果此時采用RoundRobinAssignor策略：
消費者C0：t0p0
消費者C1：t1p0
消費者C2：t1p1、t2p0、t2p1、t2p2
如果此時采用的是StickyAssignor策略：
消費者C0：t0p0
消費者C1：t1p0、t1p1
消費者C2：t2p0、t2p1、t2p2
此時消費者C0脫離了消費組，那么RoundRobinAssignor策略的分配結果為：
消費者C1：t0p0、t1p1
消費者C2：t1p0、t2p0、t2p1、t2p2
StickyAssignor策略，那么分配結果為：
消費者C1：t1p0、t1p1、t0p0
消費者C2：t2p0、t2p1、t2p2
可以看到StickyAssignor策略保留了消費者C1和C2中原有的5個分區的分配：
t1p0、t1p1、t2p0、t2p1、t2p2。
從結果上看StickyAssignor策略比另外兩者分配策略而言顯得更加的優異，這個策略的代碼實現也是異常復雜。
注意
在實際開發過程中，kafka與spark或者flink對接的較多，一個分區對應的是一個并行度，如果并行度不夠，這個時候會多個分區數據集中到一個并行度上。所以需要合理設置并行度
HBase分區
HBase每張表在底層存儲上是由至少一個Region組成，Region實際上就是HBase表的分區。HBase新建一張表時默認Region即分區的數量為1，一般在生產環境中我們都會手動給Table提前做 “預分區”，使用合適的分區策略創建好一定數量的分區并使分區均勻分布在不同regionserver上。一個分區在達到一定大小時會自動Split，一分為二
HBase分區過多有哪些影響：
- 頻繁刷寫：我們知道Region的一個列族對應一個MemStore，假設HBase表都有統一的1個列族配置，則每個Region只包含一個MemStore。通常HBase的一個MemStore默認大小為128 MB，見參數hbase.hregion.memstore.flush.size。當可用內存足夠時，每個MemStore可以分配128 MB空間。當可用內存緊張時，假設每個Region寫入壓力相同，則理論上每個MemStore會平均分配可用內存空間。因此，當節點Region過多時，每個MemStore分到的內存空間就會很小。這個時候，寫入很小的數據量就會被強制Flush到磁盤，將會導致頻繁刷寫。頻繁刷寫磁盤，會對集群HBase與HDFS造成很大的壓力，可能會導致不可預期的嚴重后果。
- 壓縮風暴：因Region過多導致的頻繁刷寫，將在磁盤上產生非常多的HFile小文件，當小文件過多的時候HBase為了優化查詢性能就會做Compaction操作，合并HFile減少文件數量。當小文件一直很多的時候，就會出現 “壓縮風暴”。Compaction非常消耗系統io資源，還會降低數據寫入的速度，嚴重的會影響正常業務的進行。
- MSLAB內存消耗較大：MSLAB(MemStore-local allocation buffer)存在于每個MemStore中，主要是為了解決HBase內存碎片問題，默認會分配 2 MB 的空間用于緩存最新數據。如果Region數量過多，MSLAB總的空間占用就會比較大。比如當前節點有1000個包含1個列族的Region，MSLAB就會使用1.95GB的堆內存，即使沒有數據寫入也會消耗這么多內存。
- Master assign region時間較長：HBase Region過多時Master分配Region的時間將會很長。特別體現在重啟HBase時Region上線時間較長，嚴重的會達到小時級，造成業務長時間等待的后果。
- 影響MapReduce并發數：當使用MapReduce操作HBase時，通常Region數量就是MapReduce的任務數，Region數量過多會導致并發數過多，產生過多的任務。任務太多將會占用大量資源，當操作包含很多Region的大表時，占用過多資源會影響其他任務的執行。
具體計算HBase合理分區數量
- ((RS memory) * (total memstore fraction)) / ((memstore size)*(column families))
| 字段 | 解釋 |
|---|---|
| RS memory | 表示regionserver堆內存大小，即HBASE_HEAPSIZE |
| total memstore fraction | 表示所有MemStore占HBASE_HEAPSIZE的比例，HBase0.98版本以后由hbase.regionserver.global.memstore.size參數控制，老版本由hbase.regionserver.global.memstore.upperLimit參數控制，默認值0.4 |
| memstore size | 即每個MemStore的大小，原生HBase中默認128M |
| column families | 即表的列族數量，通常情況下只設置1個，最多不超過3個 |
假如一個集群中每個regionserver的堆內存是32GB，那么節點上最理想的Region數量應該是32768*0.4/128 ≈ 102，所以，當前環境中單節點理想情況下大概有102個Region 最理想情況是假設每個Region上的填充率都一樣，包括數據寫入的頻次、寫入數據的大小，但實際上每個Region的負載各不相同，可能有的Region特別活躍負載特別高，有的Region則比較空閑。所以，通常我們認為2-3倍的理想Region數量也是比較合理的，針對上面舉例來說，大概200-300個Region算是合理的。
如果實際的Region數量比2~3倍的計算值還要多，就要實際觀察Region的刷寫、壓縮情況了，Region越多則風險越大。經驗告訴我們，如果單節點Region數量過千，集群可能存在較大風險
Kudu分區
為了提供可擴展性，Kudu 表被劃分為稱為 tablets 的單元，并分布在許多 tablet servers 上。行總是屬于單個 tablet 。將行分配給 tablet 的方法由在表創建期間設置的表的分區決定。kudu提供了3種分區方式：
- Range Partitioning(范圍分區) 范圍分區可以根據存入數據的數據量，均衡的存儲到各個機器上，防止機器出現負載不均衡現象
- create table people(id Type.INT32, name Type.STRING , age Type.INT32)
- RANGE (age) (
- PARTITION 0 <= VALUES < 10,
- PARTITION 10 <= VALUES < 20,
- PARTITION 20 <= VALUES < 30,
- PARTITION 30 <= VALUES < 40,
- PARTITION 40 <= VALUES < 50,
- PARTITION 50 <= VALUES < 60,
- PARTITION 60 <= VALUES < 70,
- PARTITION 70 <= VALUES < 80,
- PARTITION 80 <= VALUES < 120
- )
- Hash Partitioning(哈希分區) 哈希分區通過哈希值將行分配到許多 buckets ( 存儲桶 )之一;哈希分區是一種有效的策略，當不需要對表進行有序訪問時。哈希分區對于在 tablet 之間隨機散布這些功能是有效的，這有助于減輕熱點和 tablet 大小不均勻。
- create table rangeTable(id Type.INT32, name Type.STRING , age Type.INT32)
- HASH (id) PARTITIONS 5,
- RANGE (id) (
- PARTITION UNBOUNDED
- )
- Multilevel Partitioning(多級分區)
- create table rangeTable(id Type.INT32, name Type.STRING , age Type.INT32)
- HASH (age) PARTITIONS 5,
- RANGE (age) (
- PARTITION 0 <= VALUES < 10,
- PARTITION 10 <= VALUES < 20,
- PARTITION 20 <= VALUES < 30,
- PARTITION 30 <= VALUES < 40,
- PARTITION 40 <= VALUES < 50,
- PARTITION 50 <= VALUES < 60,
- PARTITION 60 <= VALUES < 70,
- PARTITION 70 <= VALUES < 80,
- PARTITION 80 <= VALUES < 120
哈希分區有利于最大限度地提高寫入吞吐量，而范圍分區可避免 tablet 無限增長的問題;hash分區和range分區結合，可以極大提升kudu性能
總結
優秀的設計思想需要深入的研究與發現。工作實踐總結是知識積累的很好途徑。
