女人自慰AV免费观看内涵网,日韩国产剧情在线观看网址,神马电影网特片网,最新一级电影欧美,在线观看亚洲欧美日韩,黄色视频在线播放免费观看,ABO涨奶期羡澄,第一导航fulione,美女主播操b

0
  • 聊天消息
  • 系統(tǒng)消息
  • 評論與回復
登錄后你可以
  • 下載海量資料
  • 學習在線課程
  • 觀看技術視頻
  • 寫文章/發(fā)帖/加入社區(qū)
會員中心
創(chuàng)作中心

完善資料讓更多小伙伴認識你,還能領取20積分哦,立即完善>

3天內不再提示

分析Hive與Spark分區(qū)策略的異同點

OSC開源社區(qū) ? 來源:vivo互聯(lián)網(wǎng)技術 ? 2023-04-11 09:03 ? 次閱讀

隨著技術的不斷的發(fā)展,大數(shù)據(jù)領域對于海量數(shù)據(jù)的存儲和處理的技術框架越來越多。在離線數(shù)據(jù)處理生態(tài)系統(tǒng)最具代表性的分布式處理引擎當屬Hive和Spark,它們在分區(qū)策略方面有著一些相似之處,但也存在一些不同之處。

一、概述

隨著技術的不斷的發(fā)展,大數(shù)據(jù)領域對于海量數(shù)據(jù)的存儲和處理的技術框架越來越多。在離線數(shù)據(jù)處理生態(tài)系統(tǒng)最具代表性的分布式處理引擎當屬Hive和Spark,它們在分區(qū)策略方面有著一些相似之處,但也存在一些不同之處。本篇文章將分析Hive與Spark分區(qū)策略的異同點、它們各自的優(yōu)缺點,以及一些優(yōu)化措施。

二、Hive和Spark分區(qū)概念

在了解Hive和Spark分區(qū)內容之前,首先,我們先來回顧一下Hive和Spark的分區(qū)概念。在Hive中,分區(qū)是指將表中的數(shù)據(jù)劃分為不同的目錄或者子目錄,這些目錄或子目錄的名稱通常與表的列名相關聯(lián)。比如,一個名為“t_orders_name”的表可以按照日期分為多個目錄,每個目錄名稱對應一個日期值。這樣做的好處是可以大大提高查詢效率,因為只有涉及到特定日期的查詢才需要掃描對應的目錄,而不需要去掃描整個表。Spark的分區(qū)概念與Hive類似,但是有一些不同之處,我們將在后文中進行討論。

在Hive中,分區(qū)可以基于多個列進行,這些列的值組合形成目錄名稱。例如,如果我們將“t_orders_name”表按照日期和地區(qū)分區(qū),那么目錄的名稱將包含日期和地區(qū)值的組合。在Hive中,數(shù)據(jù)存儲在分區(qū)的目錄下,而不是存儲在表的目錄下。這使得Hive可以快速訪問需要的數(shù)據(jù),而不必掃描整個表。另外,Hive的分區(qū)概念也可以用于數(shù)據(jù)分桶,分桶是將表中的數(shù)據(jù)劃分為固定數(shù)量的桶,每個桶包含相同的行。

而與Hive不同的是,Spark的分區(qū)是將數(shù)據(jù)分成小塊以便并行計算處理。在Spark中,分區(qū)的數(shù)量由Spark執(zhí)行引擎根據(jù)數(shù)據(jù)大小和硬件資源自動計算得出。Spark的分區(qū)數(shù)越多,可以并行處理的數(shù)據(jù)也就越多,因此也能更快的完成計算任務。但是,如果分區(qū)數(shù)太多,將會導致過多的任務調度和數(shù)據(jù)傳輸開銷,從而降低整體的性能。因此,Spark分區(qū)數(shù)的選擇應該考慮數(shù)據(jù)大小、硬件資源和計算任務復雜度等因素。

三、Hive和Spark分區(qū)的應用場景

在了解Hive和Spark的分區(qū)概念之后,接下來,我們來看看Hive和Spark分區(qū)在不同的應用場景中有哪些不同的優(yōu)勢。

3.1 Hive分區(qū)

Hive分區(qū)適用于大數(shù)據(jù)場景,可以對數(shù)據(jù)進行多級分區(qū),以便更細粒度地劃分數(shù)據(jù),提高查詢效率。例如,在游戲平臺的充值數(shù)據(jù)中,可以按照道具購買日期、道具付款狀態(tài)、游戲用戶ID等多個維度進行分區(qū)。這樣可以方便的進行數(shù)據(jù)統(tǒng)計、分析和查詢操作,同時避免單一分區(qū)數(shù)據(jù)過大導致的性能問題。

3.2 Spark分區(qū)

Spark分區(qū)適用于大規(guī)模數(shù)據(jù)處理場景,可以充分利用集群資源進行并行計算處理。比如,在機器學習算法的訓練過程中,可以將大量數(shù)據(jù)進行分區(qū),然后并行處理每個分區(qū)的數(shù)據(jù),從而提高算法的訓練速度和效率。另外,Spark的分布式計算引擎也可以支持在多個節(jié)點上進行數(shù)據(jù)分區(qū)和計算,從而提高整個集群的計算能力和效率。

簡而言之,Hive和Spark分區(qū)在大數(shù)據(jù)處理和分布式計算場景這都有廣泛的應用,可以通過選擇合適的分區(qū)策略和優(yōu)化措施,進一步提高數(shù)據(jù)處理的效率和性能。

四、如何選擇分區(qū)策略

在熟悉了Hive和Spark的分區(qū)概念以及應用場景后。接下來,我們來看看在Hive和Spark中如何選擇分區(qū)策略。分區(qū)策略的選擇對數(shù)據(jù)處理的效率和性能有著重要的影響。下面將分別闡述Hive和Spark分區(qū)策略的優(yōu)缺點以及如何選擇分區(qū)策略。

4.1 Hive分區(qū)策略

優(yōu)點:

Hive的分區(qū)策略可以提高查詢效率和數(shù)據(jù)處理性能,特別是在大數(shù)據(jù)集上表現(xiàn)突出。另外,Hive還支持多級分區(qū),允許更細粒度的數(shù)據(jù)劃分。

缺點:

在Hive中,分區(qū)是以目錄的形式存在的,這會導致大量的目錄和子目錄,如果分區(qū)過多,將會占用過多的存儲空間。此外,Hive的分區(qū)策略需要在創(chuàng)建表時進行設置,如果數(shù)據(jù)分布出現(xiàn)變化,需要重新設置分區(qū)策略。

4.2 Spark分區(qū)策略

優(yōu)點:

Spark的分區(qū)策略可以根據(jù)數(shù)據(jù)大小和硬件資源自動計算分區(qū)數(shù),這使得計算任務可以并行計算處理,從而提高了處理效率和性能。

缺點:

如果分區(qū)數(shù)設置不當,將會導致過多的任務調度和數(shù)據(jù)傳輸開銷,從而影響整體性能。此外,Spark的分區(qū)策略也需要根據(jù)數(shù)據(jù)大小、硬件資源和計算任務復雜度等因素進行調整。

4.3 分區(qū)策略選擇

在實際項目開發(fā)使用中,選擇合適的分區(qū)策略可以顯著提高數(shù)據(jù)處理的效率和性能。但是,如何選擇分區(qū)策略需要根據(jù)具體情況進行考慮,這里總結了一些分區(qū)策略選擇的場景:

數(shù)據(jù)集大小:如果數(shù)據(jù)集較大,可以考慮使用Hive的多級劃分策略,以便更細粒度的劃分數(shù)據(jù),提高查詢效率。如果數(shù)據(jù)集較小,可以使用Spark自動計算分區(qū)策略,以便充分利用硬件資源并提高計算效率。

計算任務復雜度:如果計算任務比較復雜,例如需要進行多個JOIN操作,可以使用Hive的分桶策略,以便加快數(shù)據(jù)訪問速度,減少JOIN操作的開銷。

硬件資源:分區(qū)策略的選擇也需要考慮硬件資源的限制。如果硬件資源比較充足,可以增加分區(qū)數(shù)以提高計算效率。如果硬件資源比較緊張,需要減少分區(qū)數(shù)以避免任務調度和數(shù)據(jù)傳輸?shù)拈_銷。

綜上所述,選擇合適的分區(qū)策略需要根據(jù)具體的情況進行考慮,包括數(shù)據(jù)集大小、計算任務復雜度和硬件資源等因素。在實際使用中,可以通過實驗和調試來找到最佳的分區(qū)策略。

五、如何優(yōu)化分區(qū)性能

除了選擇合適的分區(qū)策略之外,還可以通過一些優(yōu)化措施來進一步提高分區(qū)的性能。在Spark中,大多數(shù)的Spark任務可以通過三個階段來表述,它們分別是讀取輸入數(shù)據(jù)、使用Spark處理、保持輸出數(shù)據(jù)。Spark雖然實際數(shù)據(jù)處理主要發(fā)生在內存中,但是Spark使用的是存儲在HDFS上的數(shù)據(jù)來作為輸入和輸出,任務的調度執(zhí)行會使用大量的 I/O,存在性能瓶頸。

而Hive分區(qū)數(shù)據(jù)是存儲在HDFS上的,然而HDFS對于大量小文件支持不太友好,因為在每個NameNode內存中每個文件大概有150字節(jié)的存儲開銷,而整個HDFS集群的IOPS數(shù)量是有上限的。當文件寫入達到峰值時,會對HDFS集群的基礎架構的某些部分產(chǎn)生性能瓶頸。

5.1 通過減少 I/O 帶寬來優(yōu)化性能

在Hadoop集群中,它依靠大規(guī)模并行 I/O 來支持數(shù)千個并發(fā)任務。比如現(xiàn)有一個大小為96TB的數(shù)據(jù)節(jié)點,磁盤的大小有兩種,它們分別是8TB和16TB。具有8TB磁盤的數(shù)據(jù)節(jié)點有12塊這樣的磁盤,而具有16TB磁盤的數(shù)據(jù)節(jié)點有6塊這樣的磁盤。我們可以假設每個磁盤的平均讀寫吞吐量約為100MB/s,而這兩種不同的磁盤分布,它們對應的帶寬和IOPS,具體詳情如下表所示:

ea560c18-d7fb-11ed-bfe3-dac502259ad0.jpg

5.2 通過設置參數(shù)來優(yōu)化性能

在Hadoop集群中,每個數(shù)據(jù)節(jié)點為每個卷運行一個卷掃描器,用于掃描塊的狀態(tài)。由于卷掃描器與應用程序競爭磁盤資源,因此限制其磁盤帶寬很重要。配置 dfs.block.scanner.volume.bytes.per.second 屬性值來定義卷掃描器每秒可以掃描的字節(jié)數(shù),默認為1MB/s。

比如設置帶寬為5MB/s,掃描12TB所需要的時間為

12TB / 5MBps = (12 * 1024 * 1024 / (3600 * 24)) = 29.13天。

5.3 通過優(yōu)化Spark處理分區(qū)任務來提升性能

假如,現(xiàn)在需要重新計算歷史分區(qū)的數(shù)據(jù)表,這種場景通常用于修復錯誤或者數(shù)據(jù)質量問題。在處理包含一年數(shù)據(jù)的大型數(shù)據(jù)集(比如1TB以上)時,可能會將數(shù)據(jù)分成幾千個Spark分區(qū)來進行處理。雖然,從表面上看,這種處理方法并不是最合適的,使用動態(tài)分區(qū)并將數(shù)據(jù)結果寫入按照日期分區(qū)的Hive表中將產(chǎn)生多達上百萬個文件。

下面,我們將任務分區(qū)數(shù)縮小,現(xiàn)有一個包含3個分區(qū)的Spark任務,并且想將數(shù)據(jù)寫入到包含3個分區(qū)的Hive表。在這種情況下,希望發(fā)送的是將3個文件寫入到HDFS中,所有數(shù)據(jù)都存儲在每個分區(qū)的單個文件中。最終會生成9個文件,并且每個文件都有1個記錄。使用動態(tài)分區(qū)寫入Hive表時,每個Spark分區(qū)都由執(zhí)行程序來并行處理。

處理Spark分區(qū)數(shù)據(jù)時,每次執(zhí)行程序在給定的Spark分區(qū)中遇到新的分區(qū)時,它都會打開一個新文件。默認情況下,Spark對數(shù)據(jù)會使用Hash或者Round Robin分區(qū)器。當應用于任意數(shù)據(jù)時,可以假設這兩種方法在整個Spark分區(qū)中相對均勻且隨機分布數(shù)據(jù)。如下圖所示:

ea67cafc-d7fb-11ed-bfe3-dac502259ad0.png

理想情況下,目標文件大小應該大約是HDFS塊大小的倍數(shù),默認情況下是128MB。在Hive中,提供了一些配置參數(shù)來自動將結果寫入到合理大小的文件中,從開發(fā)者的角度來看幾乎是透明的,比如設置屬性 hive.merge.smallfiles.avgsize 和

hive.merge.size.per.task 。但是,Spark中不存在此類功能,因此,我們需要自己開發(fā)實現(xiàn),來確定一個數(shù)據(jù)集,應該寫入多少文件。

5.3.1 基于大小的計算

理論上,這是最直接的方法,設置目標大小,估算數(shù)據(jù)的大小,然后進行劃分。但是,在很多情況下,文件被寫入磁盤時會進行壓縮,并且其格式與存儲在 Java 堆中的記錄格式有所不同。這意味著估算寫入磁盤時內存的記錄大小不是一件容易的事情。雖然可以使用 Spark SizeEstimator應用程序通過內存中的數(shù)據(jù)的大小進行估算。但是,SizeEstimator會考慮數(shù)據(jù)幀、數(shù)據(jù)集的內部消耗,以及數(shù)據(jù)的大小。總體來說,這種方式不太容易準確實現(xiàn)。

5.3.2 基于行數(shù)的計算

這種方法是設置目標行數(shù),計算數(shù)據(jù)集的大小,然后執(zhí)行除法來估算目標。我們的目標行數(shù)可以通過多種方式確定,或者通過為所有數(shù)據(jù)集選擇一個靜態(tài)數(shù)字,或者通過確定磁盤上單個記錄的大小并執(zhí)行必要的計算。哪種方式最優(yōu),取決于你的數(shù)據(jù)集數(shù)量及其復雜性。計算相對來說成本較低,但是需要在計算前緩存以避免重新計算數(shù)據(jù)集。

5.3.3 靜態(tài)文件計算

最簡單的解決方案是,只要求開發(fā)者在每個寫入任務的基礎上,告訴Spark總共應該寫入多少個文件。這種方式需要給開發(fā)者一些其他方法來獲取具體的數(shù)字,可以通過這種方式來替代昂貴的計算。

5.4. 優(yōu)化Spark分發(fā)數(shù)據(jù)方式來提升性能

即使我們知道了如何將文件寫入磁盤,但是,我們仍須讓Spark以符合實際的方式來構建我們的分區(qū)。在Spark中,它提供了許多工具來確定數(shù)據(jù)在整個分區(qū)中的分布方式。但是,各種功能中隱藏著很多復雜性,在某些情況下,它們的含義并不明顯,下面將介紹Spark提供的一些選項來控制Spark輸出文件的數(shù)量。

5.4.1 合并

Spark Coalesce是一個特殊版本的重新分區(qū),它只允許減少總的分區(qū),但是不需要完全的Shuffle,因此比重新分區(qū)要快得多。它通過有效的合并分區(qū)來實現(xiàn)這一點。如下圖所示:

ea7a918c-d7fb-11ed-bfe3-dac502259ad0.png

Coalesce在某些情況下看起來是不錯的,但是也有一些問題。首先,Coalesce有一個難以使用的行為,以一個非常基礎的Spark應用程序為例,代碼如下所示:

Spark

load().map(…).filter(…).save()

比如,設置的并行度為1000,但是最終只想寫入10個文件,可以設置如下:

Spark

load().map(…).filter(…).coalesce(10).save()

但是,Spark會盡可能早的有效的將合并操作下推,因此這將執(zhí)行為如下代碼:

Spark

load().coalesce(10).map(…).filter(…).save()

有效的解決這種問題的方法是在轉換和合并之間強制執(zhí)行,代碼如下所示:

Spark

val df = load().map(…).filter(…).cache()
df.count()
df.coalesce(10)

在Spark中,緩存是必須的,否則,你將不得不重新計算數(shù)據(jù),這可能會重新消耗計算資源。然后,緩存是需要消費一定資源的,如果你的數(shù)據(jù)集無法放入內存中,或者無法釋放內存,將數(shù)據(jù)有效的存儲在內存中兩次,那么必須使用磁盤緩存,這有其自身的局限性和顯著的性能損失。

此外,正如我們看到的,通常需要執(zhí)行Shuffle來獲得我們想要的更復雜的數(shù)據(jù)集結果。因此,Coalesce僅適用于特定的情況,比如如下場景:

保證只寫入一個Hive分區(qū);

目標文件數(shù)少于你用于處理數(shù)據(jù)的Spark分區(qū)數(shù);

有充足的緩存資源。

5.4.2 簡單重新分區(qū)

在Spark中,一個簡單的重新分區(qū),可以通過設置參數(shù)來實現(xiàn),比如df.repartition(100)。在這種情況下,使用循環(huán)分區(qū)器,這意味著唯一的保證是輸出數(shù)據(jù)具有大致相同大小的Spark分區(qū),這種分區(qū)僅適用于以下情況:

保證只需要寫入一個Hive分區(qū);

正在寫入的文件數(shù)大于你的Spark分區(qū)數(shù),或者由于某些原因你無法使用合并。

5.4.3 按列重新分區(qū)

按列重新分區(qū)接收目標Spark分區(qū)計數(shù),以及要重新分區(qū)的列序列,例如,df.repartition(100,$"date")。這對于強制要求Spark將具有相同鍵的數(shù)據(jù),分發(fā)到同一個分區(qū)很有用。一般來說,這對許多Spark操作(比如JOIN)很有用。

按列重新分區(qū)使用HashPartitioner,將具有相同值的數(shù)據(jù),分發(fā)給同一個分區(qū),實際上,它將執(zhí)行以下操作:

ea8c8fe0-d7fb-11ed-bfe3-dac502259ad0.png

但是,這種方法只有在每個分區(qū)鍵都可以安全的寫入到一個文件時才有效。這是因為無論有多少特定的Hash值,它們最終都會在同一個分區(qū)中。按列重新分區(qū)僅在你寫入一個或者多個小的Hive分區(qū)時才有效。在任何其他情況下,它都是無效的,因為每個Hive分區(qū)最終都會生成一個文件,僅適用于最小的數(shù)據(jù)集。

5.4.4 按具有隨機因子的列重新分區(qū)

我們可以通過添加約束的隨機因子來按列修改重新分區(qū),具體代碼如下:

Spark

df
.withColumn("rand", rand() % filesPerPartitionKey)
.repartition(100, $"key", $"rand")

理論上,只要滿足以下條件,這種方法應該會產(chǎn)生排序規(guī)則的數(shù)據(jù)和大小均勻的文件:

Hive分區(qū)的大小大致相同;

知道每個Hive分區(qū)的目標文件數(shù)并且可以在運行時對其進行編碼。

但是,即使我們滿足上述這些條件,還有另外一個問題:散列沖突。假設,現(xiàn)在正在處理一年的數(shù)據(jù),日期作為分區(qū)的唯一鍵。如果每個分區(qū)需要5個文件,可以執(zhí)行如下代碼操作:

Spark

df.withColumn("rand", rand() % 5).repartition(5*365, $"date", $"rand")

在后臺,Scala將構造一個包含日期和隨機因子的鍵,例如(,<0-4>)。然后,如果我們查看HashPartitioner代碼,可以發(fā)現(xiàn)它將執(zhí)行以下操作:

Spark

class HashPartitioner(partitions: Int) extends Partitioner {
    def getPartition(key: Any): Int = key match {
        case null => 0
        case _ => Utils.nonNegativeMod(key.hashCode, numPartitions)
    }
}

實際上,這里面所做的事情,就是獲取關鍵元組的散列,然后使用目標數(shù)量的Spark分區(qū)獲取它的mod。我們可以分析一下在這種情況下我們的數(shù)據(jù)將如何實現(xiàn)分布,具體代碼如下:

Spark

import java.time.LocalDate
 
def hashCodeTuple(one: String, two: Int, mod: Int): Int = {
 val rawMod = (one, two).hashCode % mod
 rawMod + (if (rawMod < 0) mod else 0)
}
def hashCodeSeq(one: String, two: Int, mod: Int): Int = {
 val rawMod = Seq(one, two).hashCode % mod
 rawMod + (if (rawMod < 0) mod else 0)
}
 
def iteration(numberDS: Int, filesPerPartition: Int): (Double, Double, Double) = {
  val hashedRandKeys = (0 to numberDS - 1).map(x => LocalDate.of(2019, 1, 1).plusDays(x)).flatMap(
    x => (0 to filesPerPartition - 1).map(y => hashCodeTuple(x.toString, y, filesPerPartition*numberDS))
  )
 
  hashedRandKeys.size // Number of unique keys, with the random factor
 
  val groupedHashedKeys = hashedRandKeys.groupBy(identity).view.mapValues(_.size).toSeq
 
  groupedHashedKeys.size // number of actual sPartitions used
 
  val sortedKeyCollisions = groupedHashedKeys.filter(_._2 != 1).sortBy(_._2).reverse
   
  val sortedSevereKeyCollisions = groupedHashedKeys.filter(_._2 > 2).sortBy(_._2).reverse
 
  sortedKeyCollisions.size // number of sPartitions with a hashing collision
 
  // (collisions, occurences)
  val collisionCounts = sortedKeyCollisions.map(_._2).groupBy(identity).view.mapValues(_.size).toSeq.sortBy(_._2).reverse
   
  (
    groupedHashedKeys.size.toDouble / hashedRandKeys.size.toDouble,
    sortedKeyCollisions.size.toDouble / groupedHashedKeys.size.toDouble,
  sortedSevereKeyCollisions.size.toDouble / groupedHashedKeys.size.toDouble
  )
}
 
val results = Seq(
  iteration(365, 1),
  iteration(365, 5),
  iteration(365, 10),
  iteration(365, 100),
  iteration(365 * 2, 100),
  iteration(365 * 5, 100),
  iteration(365 * 10, 100)
)
 
val avgEfficiency = results.map(_._1).sum / results.length
val avgCollisionRate = results.map(_._2).sum / results.length
val avgSevereCollisionRate = results.map(_._3).sum / results.length
 
(avgEfficiency, avgCollisionRate, avgSevereCollisionRate) // 63.2%, 42%, 12.6%

上面的腳本計算了3個數(shù)量:

效率:非空的Spark分區(qū)與輸出文件數(shù)量的比率;

碰撞率:(date,rand)的Hash值發(fā)送沖突的Spark分區(qū)的百分比;

嚴重沖突率:同上,但是此鍵上的沖突次數(shù)為3或者更多。

沖突很重要,因為它們意味著我們的Spark分區(qū)包含多個唯一的分區(qū)鍵,而我們預計每個Spark分區(qū)只有1個。我們從分析的結果可知,我們使用了63%的執(zhí)行器,并且可能會出現(xiàn)嚴重的偏差,我們將近一半的執(zhí)行正在處理比預期多2到3倍或者在某些情況下高達8倍的數(shù)據(jù)。

現(xiàn)在,有一個解決方法,即分區(qū)縮放。在之前示例中,輸出的Spark分區(qū)數(shù)量等于預期的總文件數(shù)。如果將N個對象隨機分配給N個插槽,可以預期會有多個插槽包含多個對象,并且有幾個空插槽。因此,需要解決此問題,必須要降低對象與插槽的比率。

我們通過縮放輸出分區(qū)計數(shù)來實現(xiàn)這一點,通過將輸出Spark分區(qū)數(shù)乘以一個大因子,類似于:

Spark

df
.withColumn("rand", rand() % 5)
.repartition(5*365*SCALING_FACTOR, $"date", $"rand")

具體分析代碼如下所示:

Spark

import java.time.LocalDate
 
def hashCodeTuple(one: String, two: Int, mod: Int): Int = {
 val rawMod = (one, two).hashCode % mod
 rawMod + (if (rawMod < 0) mod else 0)
}
 
def hashCodeSeq(one: String, two: Int, mod: Int): Int = {
 val rawMod = Seq(one, two).hashCode % mod
 rawMod + (if (rawMod < 0) mod else 0)
}
 
def iteration(numberDS: Int, filesPerPartition: Int, partitionFactor: Int = 1): (Double, Double, Double, Double) = {
  val partitionCount = filesPerPartition*numberDS * partitionFactor
  val hashedRandKeys = (0 to numberDS - 1).map(x => LocalDate.of(2019, 1, 1).plusDays(x)).flatMap(
    x => (0 to filesPerPartition - 1).map(y => hashCodeTuple(x.toString, y, partitionCount))
  )
   
  hashedRandKeys.size // Number of unique keys, with the random factor
 
  val groupedHashedKeys = hashedRandKeys.groupBy(identity).view.mapValues(_.size).toSeq
 
  groupedHashedKeys.size // number of unique hashes - and thus, sPartitions with > 0 records
   
  val sortedKeyCollisions = groupedHashedKeys.filter(_._2 != 1).sortBy(_._2).reverse
   
  val sortedSevereKeyCollisions = groupedHashedKeys.filter(_._2 > 2).sortBy(_._2).reverse
 
  sortedKeyCollisions.size // number of sPartitions with a hashing collision
 
  // (collisions, occurences)
  val collisionCounts = sortedKeyCollisions.map(_._2).groupBy(identity).view.mapValues(_.size).toSeq.sortBy(_._2).reverse
   
  (
    groupedHashedKeys.size.toDouble / partitionCount,
    groupedHashedKeys.size.toDouble / hashedRandKeys.size.toDouble,
    sortedKeyCollisions.size.toDouble / groupedHashedKeys.size.toDouble,
    sortedSevereKeyCollisions.size.toDouble / groupedHashedKeys.size.toDouble
  )
}
 
// With a scale factor of 1
val results = Seq(
  iteration(365, 1),
  iteration(365, 5),
  iteration(365, 10),
  iteration(365, 100),
  iteration(365 * 2, 100),
  iteration(365 * 5, 100),
  iteration(365 * 10, 100)
)
 
val avgEfficiency = results.map(_._2).sum / results.length // What is the ratio of executors / output files
val avgCollisionRate = results.map(_._3).sum / results.length // What is the average collision rate
val avgSevereCollisionRate = results.map(_._4).sum / results.length // What is the average collision rate where 3 or more hashes collide
 
(avgEfficiency, avgCollisionRate, avgSevereCollisionRate) // 63.2% Efficiency, 42% collision rate, 12.6% severe collision rate
 
iteration(365, 5, 2) // 37.7% partitions in-use, 77.4% Efficiency, 24.4% collision rate, 4.2% severe collision rate
iteration(365, 5, 5)
iteration(365, 5, 10)
iteration(365, 5, 100)

隨著我們的比例因子接近無窮大,碰撞很快接近于0,效率接近100%。但是,這會產(chǎn)生另外一個問題,即大量Spark分區(qū)輸出將為空。同時這些空的Spark分區(qū)也會帶來一些資源開銷,增加Driver的內存大小,會使我們更容易遇到,由于異常錯誤而導致分區(qū)鍵空間意外增大的問題。

這里的一個常見方法,是在使用這種方法時不顯示設置分區(qū)(默認并行度和縮放),如果不提供分區(qū)計數(shù),則依賴Spark默認的spark.default.parallelism值。雖然,通常并行度自然高于總輸出文件數(shù)(因此,隱式提供大于1 的縮放因子)。如果滿足以下條件,這種方式依然是一種有效的方法:

Hive分區(qū)的文件數(shù)大致相等;

可以確定平均分區(qū)文件數(shù)應該是多少;

大致知道唯一分區(qū)鍵的總數(shù)。

5.4.5 按范圍重新分區(qū)

按范圍重新分區(qū)是一個特列,它不使用RoundRobin和Hash Partitioner,而是使用一種特殊的方法,叫做Range Partitioner。

范圍分區(qū)器根據(jù)某些給定鍵的順序在Spark分區(qū)之間進行拆分行,但是,它不僅僅是全局排序,而且還擁有以下特性:

具有相同散列的所有記錄將在同一個分區(qū)中結束;

所有Spark分區(qū)都將有一個最小值和最大值與之關聯(lián);

最小值和最大值將通過使用采樣來檢測關鍵頻率和范圍來確定,分區(qū)邊界將根據(jù)這些估計值進行初始設置;

分區(qū)的大小不能保證完全相等,它們的相等性基于樣本的準確性,因此,預測的每個Spark分區(qū)的最小值和最大值,分區(qū)將根據(jù)需要增大或縮小來保證前兩個條件。

總而言之,范圍分區(qū)將導致Spark創(chuàng)建與請求的Spark分區(qū)數(shù)量相等的Bucket數(shù)量,然后它將這些Bucket映射到指定分區(qū)鍵的范圍。例如,如果你的分區(qū)鍵是日期,則范圍可能是(最小值2022-01-01,最大值2023-01-01)。然后,對于每條記錄,將記錄的分區(qū)鍵與存儲Bucket的最小值和最大值進行比較,并相應的進行分配。如下圖所示:

eaa3f98c-d7fb-11ed-bfe3-dac502259ad0.png

六、總結

在選擇分區(qū)策略時,需要根據(jù)具體的應用場景和需求進行選擇。常見的分區(qū)策略包括按照時間、地域、用戶ID等多個維度進行分區(qū)。在應用分區(qū)策略時,還可以通過一些優(yōu)化措施來進一步提高分區(qū)的性能和效率,例如合理設置分區(qū)數(shù)、避免過多的分區(qū)列、減少重復數(shù)據(jù)等。

總之,分區(qū)是大數(shù)據(jù)處理和分布式計算中非常重要的技術,可以幫助我們更好的管理和處理大規(guī)模的數(shù)據(jù),提高數(shù)據(jù)處理的效率和性能,進而幫助我們更好的應對數(shù)據(jù)分析和業(yè)務應用的挑戰(zhàn)。







審核編輯:劉清

聲明:本文內容及配圖由入駐作者撰寫或者入駐合作網(wǎng)站授權轉載。文章觀點僅代表作者本人,不代表電子發(fā)燒友網(wǎng)立場。文章及其配圖僅供工程師學習之用,如有內容侵權或者其他違規(guī)問題,請聯(lián)系本站處理。 舉報投訴
  • 數(shù)據(jù)存儲

    關注

    5

    文章

    996

    瀏覽量

    51599
  • IOPs
    +關注

    關注

    0

    文章

    12

    瀏覽量

    14384
  • HDFS
    +關注

    關注

    1

    文章

    31

    瀏覽量

    9824
  • SPARK
    +關注

    關注

    1

    文章

    106

    瀏覽量

    20391

原文標題:Hive和Spark分區(qū)策略剖析

文章出處:【微信號:OSC開源社區(qū),微信公眾號:OSC開源社區(qū)】歡迎添加關注!文章轉載請注明出處。

收藏 人收藏

    評論

    相關推薦
    熱點推薦

    分布式存儲數(shù)據(jù)恢復—虛擬機上hbase和hive數(shù)據(jù)庫數(shù)據(jù)恢復案例

    分布式存儲數(shù)據(jù)恢復環(huán)境: 16臺某品牌R730xd服務器節(jié)點,每臺服務器節(jié)點上有數(shù)臺虛擬機。 虛擬機上部署Hbase和Hive數(shù)據(jù)庫。 分布式存儲故障: 數(shù)據(jù)庫底層文件被誤刪除,數(shù)據(jù)庫不能使用。要求恢復hbase和hive數(shù)據(jù)庫。
    的頭像 發(fā)表于 04-17 11:05 ?150次閱讀

    NVIDIA加速的Apache Spark助力企業(yè)節(jié)省大量成本

    隨著 NVIDIA 推出 Aether 項目,通過采用 NVIDIA 加速的 Apache Spark 企業(yè)得以自動加速其數(shù)據(jù)中心規(guī)模的分析工作負載,從而節(jié)省數(shù)百萬美元。
    的頭像 發(fā)表于 03-25 15:09 ?419次閱讀
    NVIDIA加速的Apache <b class='flag-5'>Spark</b>助力企業(yè)節(jié)省大量成本

    永磁同步電機矢量控制策略分析

    本文通過矢量控制策略采用 id=0 控制方案快速準確地控制轉矩,實現(xiàn)調速系統(tǒng)具有較高的動態(tài)性能。并利用了 Matlab 工具對永磁同步電機矢量控制系統(tǒng)在空載起動、轉速突變、負載突變進行了仿真研究。 點擊附件可直接打開查看全文*附件:永磁同步電機矢量控制策略
    發(fā)表于 03-20 12:57

    Nand flash 和SD卡(SD NAND)存儲扇區(qū)分配表異同

    NAND Flash 和 SD卡(SD NAND)的存儲扇區(qū)分配表在原理上有相似之處,但由于二者的結構和應用場景不同,也存在一些差異。以下是它們的異同點和用法介紹:相同點: 基本功能:NAND
    發(fā)表于 03-13 10:45

    磁盤分區(qū)工具parted的使用方法

    傳統(tǒng)的MBR分區(qū)表格式,僅支持最大四個主分區(qū),而且不可以格式化2TB以上的磁盤,因此,大磁盤更適合使用parted工具進行GPT的分區(qū)格式。
    的頭像 發(fā)表于 03-03 09:39 ?662次閱讀

    電源反接制動和倒拉反接制動有何異同點

    的應用場景:? ? ??●?需要在大負載下快速停止的場合。? ? ??●?需要精確控制停止位置的場合。? ? ??異同點分析? ? ??相同點:? ? ??1、目的相同 :都是為了實現(xiàn)電機的快速停止
    的頭像 發(fā)表于 10-23 15:52 ?1014次閱讀

    Linux磁盤分區(qū)擴容方法

    linux分區(qū)常用命令:fdisk,修改MBR分區(qū)表,MBR格式,被修改的分區(qū)大小最大為2T。
    的頭像 發(fā)表于 10-23 11:46 ?918次閱讀
    Linux磁盤<b class='flag-5'>分區(qū)</b>擴容方法

    使用FAL分區(qū)管理與easyflash變量管理

    1.FAL組件1.1什么是FALFAL(FlashAbstractionLayer)Flash抽象層,是對Flash及基于Flash的分區(qū)進行管理、操作的抽象層,對上層統(tǒng)一了Flash及分區(qū)操作
    的頭像 發(fā)表于 10-01 08:10 ?1657次閱讀
    使用FAL<b class='flag-5'>分區(qū)</b>管理與easyflash變量管理

    大數(shù)據(jù)從業(yè)者必知必會的Hive SQL調優(yōu)技巧

    大數(shù)據(jù)從業(yè)者必知必會的Hive SQL調優(yōu)技巧 摘要 :在大數(shù)據(jù)領域中,Hive SQL被廣泛應用于數(shù)據(jù)倉庫的數(shù)據(jù)查詢和分析。然而,由于數(shù)據(jù)量龐大和復雜的查詢需求,Hive SQL查詢
    的頭像 發(fā)表于 09-24 13:30 ?598次閱讀

    電源反接制動和倒拉反接制動有何異同點

    電源反接制動和倒拉反接制動是兩種電機制動方式,它們在工業(yè)自動化和電機控制領域中有著廣泛的應用。這兩種制動方式各有特點和適用場景,下面我將介紹它們的異同點。 電源反接制動 電源反接制動是一種電機
    的頭像 發(fā)表于 09-19 09:10 ?2001次閱讀

    RK3568修改eMMC分區(qū)大小

    我司IAC-RK3568-CM根據(jù)eMMC大小的不同,有著不同規(guī)格的產(chǎn)品,不論eMMC大小如何改變,其分區(qū)的配置大同小異,除了eMMC廠商所使用的區(qū)域無法變更外,留給用戶的區(qū)域可自由寫入。我司產(chǎn)品
    的頭像 發(fā)表于 09-13 08:00 ?1188次閱讀
    RK3568修改eMMC<b class='flag-5'>分區(qū)</b>大小

    spark為什么比mapreduce快?

    spark為什么比mapreduce快? 首先澄清幾個誤區(qū): 1:兩者都是基于內存計算的,任何計算框架都肯定是基于內存的,所以網(wǎng)上說的spark是基于內存計算所以快,顯然是錯誤的 2;DAG計算模型
    的頭像 發(fā)表于 09-06 09:45 ?448次閱讀

    spark運行的基本流程

    前言: 由于最近對spark的運行流程非常感興趣,所以閱讀了《Spark大數(shù)據(jù)處理:技術、應用與性能優(yōu)化》一書。通過這本書的學習,了解了spark的核心技術、實際應用場景以及性能優(yōu)化的方法。本文旨在
    的頭像 發(fā)表于 07-02 10:31 ?643次閱讀
    <b class='flag-5'>spark</b>運行的基本流程

    Spark基于DPU的Native引擎算子卸載方案

    1.背景介紹 Apache Spark(以下簡稱Spark)是一個開源的分布式計算框架,由UC Berkeley AMP Lab開發(fā),可用于批處理、交互式查詢(Spark SQL)、實時流處理
    的頭像 發(fā)表于 06-28 17:12 ?957次閱讀
    <b class='flag-5'>Spark</b>基于DPU的Native引擎算子卸載方案

    Spark+Hive”在DPU環(huán)境下的性能測評 | OLAP數(shù)據(jù)庫引擎選型白皮書(24版)DPU部分節(jié)選

    在奇點云2024年版《OLAP數(shù)據(jù)庫引擎選型白皮書》中,中科馭數(shù)聯(lián)合奇點云針對Spark+Hive這類大數(shù)據(jù)計算場景下的主力引擎,測評DPU環(huán)境下對比CPU環(huán)境下的性能提升效果。特此節(jié)選該章節(jié)內容,與大家共享。
    的頭像 發(fā)表于 05-30 16:09 ?769次閱讀
    “<b class='flag-5'>Spark+Hive</b>”在DPU環(huán)境下的性能測評 | OLAP數(shù)據(jù)庫引擎選型白皮書(24版)DPU部分節(jié)選