大數(shù)據(jù)分布式中各個框架總結(jié)

前言在大數(shù)據(jù)分布式中，分區(qū)，分桶，分片是設(shè)計框架的重點。此篇就來總結(jié)各個框架。建議收藏

目錄

Hive分區(qū)與分桶

ES分片

Kafka分區(qū)

HBase分區(qū)

Kudu分區(qū)

HiveHive分區(qū)

是按照數(shù)據(jù)表的某列或者某些列分為多區(qū)，在hive存儲上是hdfs文件，也就是文件夾形式。現(xiàn)在最常用的跑T+1數(shù)據(jù)，按當(dāng)天時間分區(qū)的較多。

把每天通過sqoop或者datax拉取的一天的數(shù)據(jù)存儲一個區(qū)，也就是所謂的文件夾與文件。在查詢時只要指定分區(qū)字段的值就可以直接從該分區(qū)查找即可。創(chuàng)建分區(qū)表的時候，要通過關(guān)鍵字 partitioned by （column name string）聲明該表是分區(qū)表，并且是按照字段column name進(jìn)行分區(qū)，column name值一致的所有記錄存放在一個分區(qū)中，分區(qū)屬性name的類型是string類型。

當(dāng)然，可以依據(jù)多個列進(jìn)行分區(qū)，即對某個分區(qū)的數(shù)據(jù)按照某些列繼續(xù)分區(qū)。

向分區(qū)表導(dǎo)入數(shù)據(jù)的時候，要通過關(guān)鍵字partition（（column name=“xxxx”）顯示聲明數(shù)據(jù)要導(dǎo)入到表的哪個分區(qū)

設(shè)置分區(qū)的影響

首先是hive本身對分區(qū)數(shù)有限制，不過可以修改限制的數(shù)量；

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對單個目錄下的目錄數(shù)量或者文件數(shù)量也是有限制的，也是可以修改的；

NN的內(nèi)存肯定會限制，這是最重要的，如果分區(qū)數(shù)很大，會影響NN服務(wù)，進(jìn)而影響一系列依賴于NN的服務(wù)。所以最好合理設(shè)置分區(qū)規(guī)則，對小文件也可以定期合并，減少NN的壓力。

Hive分桶

在分區(qū)數(shù)量過于龐大以至于可能導(dǎo)致文件系統(tǒng)崩潰時，我們就需要使用分桶來解決問題

分桶是相對分區(qū)進(jìn)行更細(xì)粒度的劃分。分桶則是指定分桶表的某一列，讓該列數(shù)據(jù)按照哈希取模的方式隨機(jī)、均勻的分發(fā)到各個桶文件中。因為分桶操作需要根據(jù)某一列具體數(shù)據(jù)來進(jìn)行哈希取模操作，故指定的分桶列必須基于表中的某一列（字段）要使用關(guān)鍵字clustered by 指定分區(qū)依據(jù)的列名，還要指定分為多少桶

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分區(qū)分桶區(qū)別

分區(qū)是表的部分列的集合，可以為頻繁使用的數(shù)據(jù)建立分區(qū)，這樣查找分區(qū)中的數(shù)據(jù)時就不需要掃描全表，這對于提高查找效率很有幫助

不同于分區(qū)對列直接進(jìn)行拆分，桶往往使用列的哈希值對數(shù)據(jù)打散，并分發(fā)到各個不同的桶中從而完成數(shù)據(jù)的分桶過程

分區(qū)和分桶最大的區(qū)別就是分桶隨機(jī)分割數(shù)據(jù)庫，分區(qū)是非隨機(jī)分割數(shù)據(jù)庫

ElasticSearch分片主分片：用于解決數(shù)據(jù)水平擴(kuò)展的問題，一個索引的所有數(shù)據(jù)是分布在所有主分片之上的（每個主分片承擔(dān)一部分?jǐn)?shù)據(jù)，主分片又分布在不同的節(jié)點上），一個索引的主分片數(shù)量只能在創(chuàng)建時指定，后期無法修改，除非對數(shù)據(jù)進(jìn)行重新構(gòu)建索引（reindex操作）。

副本分片：用于解決數(shù)據(jù)高可用的問題，一個副本分片即一個主分片的拷貝，其數(shù)量可以動態(tài)調(diào)整，通過增加副本分片也可以實現(xiàn)提升系統(tǒng)讀性能的作用。

在集群中唯一一個空節(jié)點上創(chuàng)建一個叫做 blogs 的索引。默認(rèn)情況下，一個索引被分配 5 個主分片

{

“settings”： {

“number_of_shards”： 5，

“number_of_replicas”： 1

}

}

到底分配到那個shard上呢？

shard = hash（routing） % number_of_primary_shards

routing 是一個可變值，默認(rèn)是文檔的 _id ，也可以設(shè)置成一個自定義的值。routing 通過 hash 函數(shù)生成一個數(shù)字，然后這個數(shù)字再除以 number_of_primary_shards （主分片的數(shù)量）后得到余數(shù) 。這個在 0 到 number_of_primary_shards 之間的余數(shù)，就是所尋求的文檔所在分片的位置。

如果數(shù)量變化了，那么所有之前路由的值都會無效，文檔也再也找不到了

分片過少如15個節(jié)點，5個主分片，1個副本會造成每個索引最多只能使用10個節(jié)點（5個主分片，5個從分片），剩余5節(jié)點并沒有利用上；資源浪費如：3節(jié)點；3分主分片，1副本當(dāng)數(shù)據(jù)量較大的時，每個分片就會比較大

分片過多

創(chuàng)建分片慢：es創(chuàng)建分片的速度會隨著集群內(nèi)分片數(shù)的增加而變慢。

集群易崩潰：在觸發(fā)es 自動創(chuàng)建Index時，由于創(chuàng)建速度太慢，容易導(dǎo)致大量寫入請求堆積在內(nèi)存，從而壓垮集群。

寫入拒絕：分片過多的場景中，如果不能及時掌控業(yè)務(wù)的變化，可能經(jīng)常遇到單分片記錄超限、寫入拒絕等問題。

分片的注意事項

避免使用非常大的分片，因為這會對群集從故障中恢復(fù)的能力產(chǎn)生負(fù)面影響。對分片的大小沒有固定的限制，但是通常情況下很多場景限制在 30GB 的分片大小以內(nèi)。

當(dāng)在ElasticSearch集群中配置好你的索引后， 你要明白在集群運(yùn)行中你無法調(diào)整分片設(shè)置。 既便以后你發(fā)現(xiàn)需要調(diào)整分片數(shù)量， 你也只能新建創(chuàng)建并對數(shù)據(jù)進(jìn)行重新索引。

如果擔(dān)心數(shù)據(jù)的快速增長， 建議根據(jù)這條限制： ElasticSearch推薦的最大JVM堆空間 是 30~32G， 所以把分片最大容量限制為 30GB， 然后再對分片數(shù)量做合理估算。例如， 如果的數(shù)據(jù)能達(dá)到 200GB， 則最多分配7到8個分片。

kafka分區(qū)生產(chǎn)者

分區(qū)的原因

方便在集群中擴(kuò)展，每個Partition可以通過調(diào)整以適應(yīng)它所在的機(jī)器，而一個topic又可以有多個Partition組成，因此整個集群就可以適應(yīng)任意大小的數(shù)據(jù)了；

可以提高并發(fā)，因為可以以Partition為單位讀寫了。

分區(qū)的原則

指明 partition 的情況下，直接將指明的值直接作為 partiton 值；

沒有指明 partition 值但有 key 的情況下，將 key 的 hash 值與 topic 的 partition 數(shù)進(jìn)行取余得到 partition 值；

既沒有 partition 值又沒有 key 值的情況下，第一次調(diào)用時隨機(jī)生成一個整數(shù)（后面每次調(diào)用在這個整數(shù)上自增），將這個值與 topic 可用的 partition 總數(shù)取余得到 partition 值，也就是常說的 round-robin 算法。

消費者

分區(qū)分配策略

一個consumer group中有多個consumer，一個 topic有多個partition，所以必然會涉及到partition的分配問題，即確定那個partition由哪個consumer來消費Kafka有三種分配策略，一是RoundRobin，一是Range。高版本還有一個StickyAssignor策略將分區(qū)的所有權(quán)從一個消費者移到另一個消費者稱為重新平衡（rebalance）。當(dāng)以下事件發(fā)生時，Kafka 將會進(jìn)行一次分區(qū)分配：

同一個 Consumer Group 內(nèi)新增消費者

消費者離開當(dāng)前所屬的Consumer Group，包括shuts down 或 crashes

Range分區(qū)分配策略

Range是對每個Topic而言的（即一個Topic一個Topic分），首先對同一個Topic里面的分區(qū)按照序號進(jìn)行排序，并對消費者按照字母順序進(jìn)行排序。然后用Partitions分區(qū)的個數(shù)除以消費者線程的總數(shù)來決定每個消費者線程消費幾個分區(qū)。如果除不盡，那么前面幾個消費者線程將會多消費一個分區(qū)。假設(shè)n=分區(qū)數(shù)/消費者數(shù)量，m=分區(qū)數(shù)%消費者數(shù)量，那么前m個消費者每個分配n+1個分區(qū)，后面的（消費者數(shù)量-m）個消費者每個分配n個分區(qū)。假如有10個分區(qū)，3個消費者線程，把分區(qū)按照序號排列

0，1，2，3，4，5，6，7，8，9

消費者線程為

C1-0，C2-0，C2-1

那么用partition數(shù)除以消費者線程的總數(shù)來決定每個消費者線程消費幾個partition，如果除不盡，前面幾個消費者將會多消費一個分區(qū)。在我們的例子里面，我們有10個分區(qū)，3個消費者線程，10/3 = 3，而且除除不盡，那么消費者線程C1-0將會多消費一個分區(qū)，所以最后分區(qū)分配的結(jié)果看起來是這樣的：

C1-0：0，1，2，3

C2-0：4，5，6

C2-1：7，8，9

如果有11個分區(qū)將會是：

C1-0：0，1，2，3

C2-0：4，5，6，7

C2-1：8，9，10

假如我們有兩個主題T1，T2，分別有10個分區(qū)，最后的分配結(jié)果將會是這樣：

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分區(qū)分配策略

RoundRobinAssignor策略的原理是將消費組內(nèi)所有消費者以及消費者所訂閱的所有topic的partition按照字典序排序，然后通過輪詢方式逐個將分區(qū)以此分配給每個消費者。使用RoundRobin策略有兩個前提條件必須滿足：

同一個消費者組里面的所有消費者的num.streams（消費者消費線程數(shù)）必須相等；

每個消費者訂閱的主題必須相同。

加入按照 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

最后分區(qū)分配的結(jié)果為：

C1-0 將消費 T1-5， T1-2， T1-6 分區(qū)

C1-1 將消費 T1-3， T1-1， T1-9 分區(qū)

C2-0 將消費 T1-0， T1-4 分區(qū)

C2-1 將消費 T1-8， T1-7 分區(qū)

StickyAssignor分區(qū)分配策略

Kafka從0.11.x版本開始引入這種分配策略，它主要有兩個目的：

分區(qū)的分配要盡可能的均勻，分配給消費者者的主題分區(qū)數(shù)最多相差一個

分區(qū)的分配盡可能的與上次分配的保持相同。

當(dāng)兩者發(fā)生沖突時，第一個目標(biāo)優(yōu)先于第二個目標(biāo)。鑒于這兩個目的，StickyAssignor策略的具體實現(xiàn)要比RangeAssignor和RoundRobinAssignor這兩種分配策略要復(fù)雜很多。

假設(shè)消費組內(nèi)有3個消費者

C0、C1、C2

它們都訂閱了4個主題：

t0、t1、t2、t3

并且每個主題有2個分區(qū)，也就是說整個消費組訂閱了

t0p0、t0p1、t1p0、t1p1、t2p0、t2p1、t3p0、t3p1這8個分區(qū)

最終的分配結(jié)果如下：

消費者C0：t0p0、t1p1、t3p0

消費者C1：t0p1、t2p0、t3p1

消費者C2：t1p0、t2p1

這樣初看上去似乎與采用RoundRobinAssignor策略所分配的結(jié)果相同

此時假設(shè)消費者C1脫離了消費組，那么消費組就會執(zhí)行再平衡操作，進(jìn)而消費分區(qū)會重新分配。如果采用RoundRobinAssignor策略，那么此時的分配結(jié)果如下：

消費者C0：t0p0、t1p0、t2p0、t3p0

消費者C2：t0p1、t1p1、t2p1、t3p1

如分配結(jié)果所示，RoundRobinAssignor策略會按照消費者C0和C2進(jìn)行重新輪詢分配。而如果此時使用的是StickyAssignor策略，那么分配結(jié)果為：

消費者C0：t0p0、t1p1、t3p0、t2p0

消費者C2：t1p0、t2p1、t0p1、t3p1

可以看到分配結(jié)果中保留了上一次分配中對于消費者C0和C2的所有分配結(jié)果，并將原來消費者C1的“負(fù)擔(dān)”分配給了剩余的兩個消費者C0和C2，最終C0和C2的分配還保持了均衡。

如果發(fā)生分區(qū)重分配，那么對于同一個分區(qū)而言有可能之前的消費者和新指派的消費者不是同一個，對于之前消費者進(jìn)行到一半的處理還要在新指派的消費者中再次復(fù)現(xiàn)一遍，這顯然很浪費系統(tǒng)資源。StickyAssignor策略如同其名稱中的“sticky”一樣，讓分配策略具備一定的“粘性”，盡可能地讓前后兩次分配相同，進(jìn)而減少系統(tǒng)資源的損耗以及其它異常情況的發(fā)生。

到目前為止所分析的都是消費者的訂閱信息都是相同的情況，我們來看一下訂閱信息不同的情況下的處理。

舉例，同樣消費組內(nèi)有3個消費者：

C0、C1、C2

集群中有3個主題：

t0、t1、t2

這3個主題分別有

1、2、3個分區(qū)

也就是說集群中有

t0p0、t1p0、t1p1、t2p0、t2p1、t2p2這6個分區(qū)

消費者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策略的分配結(jié)果為：

消費者C1：t0p0、t1p1

消費者C2：t1p0、t2p0、t2p1、t2p2

StickyAssignor策略，那么分配結(jié)果為：

消費者C1：t1p0、t1p1、t0p0

消費者C2：t2p0、t2p1、t2p2

可以看到StickyAssignor策略保留了消費者C1和C2中原有的5個分區(qū)的分配：

t1p0、t1p1、t2p0、t2p1、t2p2。

從結(jié)果上看StickyAssignor策略比另外兩者分配策略而言顯得更加的優(yōu)異，這個策略的代碼實現(xiàn)也是異常復(fù)雜。

注意

在實際開發(fā)過程中，kafka與spark或者flink對接的較多，一個分區(qū)對應(yīng)的是一個并行度，如果并行度不夠，這個時候會多個分區(qū)數(shù)據(jù)集中到一個并行度上。所以需要合理設(shè)置并行度

HBase分區(qū)HBase每張表在底層存儲上是由至少一個Region組成，Region實際上就是HBase表的分區(qū)。HBase新建一張表時默認(rèn)Region即分區(qū)的數(shù)量為1，一般在生產(chǎn)環(huán)境中我們都會手動給Table提前做 “預(yù)分區(qū)”，使用合適的分區(qū)策略創(chuàng)建好一定數(shù)量的分區(qū)并使分區(qū)均勻分布在不同regionserver上。一個分區(qū)在達(dá)到一定大小時會自動Split，一分為二

HBase分區(qū)過多有哪些影響：

頻繁刷寫：我們知道Region的一個列族對應(yīng)一個MemStore，假設(shè)HBase表都有統(tǒng)一的1個列族配置，則每個Region只包含一個MemStore。通常HBase的一個MemStore默認(rèn)大小為128 MB，見參數(shù)hbase.hregion.memstore.flush.size。當(dāng)可用內(nèi)存足夠時，每個MemStore可以分配128 MB空間。當(dāng)可用內(nèi)存緊張時，假設(shè)每個Region寫入壓力相同，則理論上每個MemStore會平均分配可用內(nèi)存空間。因此，當(dāng)節(jié)點Region過多時，每個MemStore分到的內(nèi)存空間就會很小。這個時候，寫入很小的數(shù)據(jù)量就會被強(qiáng)制Flush到磁盤，將會導(dǎo)致頻繁刷寫。頻繁刷寫磁盤，會對集群HBase與HDFS造成很大的壓力，可能會導(dǎo)致不可預(yù)期的嚴(yán)重后果。

壓縮風(fēng)暴：因Region過多導(dǎo)致的頻繁刷寫，將在磁盤上產(chǎn)生非常多的HFile小文件，當(dāng)小文件過多的時候HBase為了優(yōu)化查詢性能就會做Compaction操作，合并HFile減少文件數(shù)量。當(dāng)小文件一直很多的時候，就會出現(xiàn) “壓縮風(fēng)暴”。Compaction非常消耗系統(tǒng)io資源，還會降低數(shù)據(jù)寫入的速度，嚴(yán)重的會影響正常業(yè)務(wù)的進(jìn)行。

MSLAB內(nèi)存消耗較大：MSLAB（MemStore-local allocation buffer）存在于每個MemStore中，主要是為了解決HBase內(nèi)存碎片問題，默認(rèn)會分配 2 MB 的空間用于緩存最新數(shù)據(jù)。如果Region數(shù)量過多，MSLAB總的空間占用就會比較大。比如當(dāng)前節(jié)點有1000個包含1個列族的Region，MSLAB就會使用1.95GB的堆內(nèi)存，即使沒有數(shù)據(jù)寫入也會消耗這么多內(nèi)存。

Master assign region時間較長：HBase Region過多時Master分配Region的時間將會很長。特別體現(xiàn)在重啟HBase時Region上線時間較長，嚴(yán)重的會達(dá)到小時級，造成業(yè)務(wù)長時間等待的后果。

影響MapReduce并發(fā)數(shù)：當(dāng)使用MapReduce操作HBase時，通常Region數(shù)量就是MapReduce的任務(wù)數(shù)，Region數(shù)量過多會導(dǎo)致并發(fā)數(shù)過多，產(chǎn)生過多的任務(wù)。任務(wù)太多將會占用大量資源，當(dāng)操作包含很多Region的大表時，占用過多資源會影響其他任務(wù)的執(zhí)行。

具體計算HBase合理分區(qū)數(shù)量

（（RS memory） * （total memstore fraction）） / （（memstore size）*（column families））

字段解釋

RS memory表示regionserver堆內(nèi)存大小，即HBASE_HEAPSIZE

total memstore fraction表示所有MemStore占HBASE_HEAPSIZE的比例，HBase0.98版本以后由hbase.regionserver.global.memstore.size參數(shù)控制，老版本由hbase.regionserver.global.memstore.upperLimit參數(shù)控制，默認(rèn)值0.4

memstore size即每個MemStore的大小，原生HBase中默認(rèn)128M

column families即表的列族數(shù)量，通常情況下只設(shè)置1個，最多不超過3個

假如一個集群中每個regionserver的堆內(nèi)存是32GB，那么節(jié)點上最理想的Region數(shù)量應(yīng)該是32768*0.4/128 ≈ 102，所以，當(dāng)前環(huán)境中單節(jié)點理想情況下大概有102個Region最理想情況是假設(shè)每個Region上的填充率都一樣，包括數(shù)據(jù)寫入的頻次、寫入數(shù)據(jù)的大小，但實際上每個Region的負(fù)載各不相同，可能有的Region特別活躍負(fù)載特別高，有的Region則比較空閑。所以，通常我們認(rèn)為2-3倍的理想Region數(shù)量也是比較合理的，針對上面舉例來說，大概200-300個Region算是合理的。

如果實際的Region數(shù)量比2~3倍的計算值還要多，就要實際觀察Region的刷寫、壓縮情況了，Region越多則風(fēng)險越大。經(jīng)驗告訴我們，如果單節(jié)點Region數(shù)量過千，集群可能存在較大風(fēng)險

Kudu分區(qū)為了提供可擴(kuò)展性，Kudu 表被劃分為稱為 tablets 的單元，并分布在許多 tablet servers 上。行總是屬于單個 tablet 。將行分配給 tablet 的方法由在表創(chuàng)建期間設(shè)置的表的分區(qū)決定。kudu提供了3種分區(qū)方式：

Range Partitioning（范圍分區(qū)）范圍分區(qū)可以根據(jù)存入數(shù)據(jù)的數(shù)據(jù)量，均衡的存儲到各個機(jī)器上，防止機(jī)器出現(xiàn)負(fù)載不均衡現(xiàn)象

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（哈希分區(qū)）哈希分區(qū)通過哈希值將行分配到許多 buckets （ 存儲桶 ）之一；哈希分區(qū)是一種有效的策略，當(dāng)不需要對表進(jìn)行有序訪問時。哈希分區(qū)對于在 tablet 之間隨機(jī)散布這些功能是有效的，這有助于減輕熱點和 tablet 大小不均勻。

create table rangeTable（id Type.INT32， name Type.STRING ， age Type.INT32）

HASH （id） PARTITIONS 5，

RANGE （id） （

PARTITION UNBOUNDED

）

Multilevel Partitioning（多級分區(qū)）

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

哈希分區(qū)有利于最大限度地提高寫入吞吐量，而范圍分區(qū)可避免 tablet 無限增長的問題；hash分區(qū)和range分區(qū)結(jié)合，可以極大提升kudu性能。

責(zé)任編輯：haq