如何根據(jù)Spark SQL explaind中的統(tǒng)計信息深入了解CBO優(yōu)化

如何根據(jù)Spark SQL explaind中的統(tǒng)計信息深入了解CBO優(yōu)化，相信很多沒有經(jīng)驗的人對此束手無策，為此本文總結了問題出現(xiàn)的原因和解決方法，通過這篇文章希望你能解決這個問題。

Spark SQL 優(yōu)化器使用兩種優(yōu)化方式：基于規(guī)則的和基于代價的。前者依賴于啟發(fā)式規(guī)則，而后者依賴于數(shù)據(jù)的統(tǒng)計屬性。在這篇文章里,我們解釋一下在底層這些統(tǒng)計信息是怎么被用到，以及哪些場景下是有用的，并且怎么來使用這些統(tǒng)計信息。
大部分基于啟發(fā)式的優(yōu)化規(guī)則都沒有考慮到被處理的數(shù)據(jù)屬性。比如：基于啟發(fā)式的PredicatePushDown規(guī)則就是基于先過濾再計算的假設。
然而有些場景spark能夠通過數(shù)據(jù)的統(tǒng)計信息來得出更好的計劃，這通常被稱作基于代價的優(yōu)化或者CBO，我們來探討一下細節(jié)。

怎么看到統(tǒng)計信息

為了能夠看到一個表的統(tǒng)計信息首先我們需要通過運行sql語句來計算(所有的SQL語句可以通過使用sql()函數(shù)來指定，spark.sql(需要指定的sql字符串))：

ANALYZE TABLE table_name COMPUTE STATISTICS

運行完這個以后，表級別的統(tǒng)計信息就會統(tǒng)計出來并且被存儲在元數(shù)據(jù)中，我們可以通過以下語句來查看：

DESCRIBE EXTENDED table_name

這將會展現(xiàn)一些表屬性以及表級別的統(tǒng)計信息。這有兩種維度信息：rowCount和sizeBytes： 除了表級別的統(tǒng)計信息，這也有列級別的統(tǒng)計信息，我們可以通過一下語句去計算和查看：

ANALYZE TABLE table_name COMPUTE STATISTICS FOR COLUMNS col_name
DESCRIBE EXTENDED table_name column_name

這將展示給我們類似一下的表(在這個例子中我們使用的列是user_id): 就像你看到的，這里有各種各樣的列維度信息,像最大最大最小值，null值的數(shù)量，去重的值的數(shù)量 (近似值)等。 從Spark 3.0以來，這里有更多的選項去展示這些信息，能夠展示的不僅僅是表也包括了實際的查詢語句?？梢酝ㄟ^explain的的mode參數(shù)來實現(xiàn)：

spark.table(table_name).explain(mode='cost')

這個將會給我們展示兩種查詢計劃，物理計劃和優(yōu)化的邏輯計劃，該邏輯計劃將會展示一些統(tǒng)計信息，就像以下圖片展示的： 這個重要的一點是你能看到計劃的每個操作的的統(tǒng)計信息，所以在各種各樣的轉變之后你能看到統(tǒng)計信息的估算。這些統(tǒng)計信息首先通過Relation操作也就是所謂的葉子節(jié)點計算出來的，并且每個葉子節(jié)點都負責計算,后續(xù)經(jīng)過一些規(guī)則通過邏輯計劃進行傳播。
接下來，我們將會了解葉子節(jié)點是這么計算統(tǒng)計信息，以及怎么傳播的。

統(tǒng)計信息怎么被計算的

葉子節(jié)點計算統(tǒng)計信息有三種方式:第一種也是最好的一種是從元數(shù)據(jù)中獲取的統(tǒng)計信息。第二種是spark將會使用InMemoryFileIndex，他將會調(diào)用底層的 Hadoop API去收集數(shù)據(jù)源中的每個文件的的大小并且求和得到總值sizeInBytes（這里只有sizeInBytes度量會被計算到），最后一種方式是spark將會使用默認的sizeInBytes維度值，該值由spark.sql.defaultSizeInBytes配置 并且該默認值為8EIB，所以基本上spark對于Relation sizeInBytes將會盡可能的進行重新計算覆蓋。（這也是只有sizeInBytes這種度量用到）,這三種方式可以通過一下圖表進行描述： 這個圖標是一個樹形，每個節(jié)點是一條件，假如條件為真，我們將轉到T，否則轉到F。葉子將會代表統(tǒng)計信息將會計算的實際方式。例如：InMemoryFI 意味著只有sizeInBytes將調(diào)用Hadoop API進行計算。另一方面，Stats From M 意味著統(tǒng)計信息將會從元數(shù)據(jù)獲得，然而在左邊的數(shù) 所有統(tǒng)計信息將會從元數(shù)據(jù)獲取，而右邊只有metricsInBytes維度將會被獲取。葉子節(jié)點CatalogFileIndex 代表著最后一種方法-默認值為8EIB的sizeInBytes將會被使用到。
在圖表中，我們有四種條件，第一種決定了統(tǒng)計信息怎么被獲取：假如我們讀取數(shù)據(jù)作為一個表df=spark.table(table_name)，那我們就進入到左邊，否則進入到右邊，下一個條件是 是否基于代價的優(yōu)化(CBO)是否開啟，這個通過spark.sql.cbo.enabled配置，默認值是false（到spark 3.0.0）.第三個條件是在元數(shù)據(jù)的統(tǒng)計信息是否通過analyzed table command（ATC）計算出來的，最后一個是表是否分區(qū)。 最好的情況是 我們讀取數(shù)據(jù)作為一個表，CBO是開啟的，而且已經(jīng)運行了ATC，這種情況下，所有的統(tǒng)計信息將會從元數(shù)據(jù)中獲?。ǔ藦膔owCount計算的sizeInBytes）,另一個方面，最壞的情況是，我們讀取數(shù)據(jù)作為一個表，但是ATC沒有運行，而且表是分區(qū)的，這種情況下默認的sizeInBytes將會從配置中讀取，并且計算是很不精確的，注意最壞的情況跟CBO是否開啟是無關的。注意一點：假如表不是分區(qū)的，spark將會使用Hadoop API計算sizeInBytes,所以表是否分區(qū)直接影響了葉子節(jié)點的統(tǒng)計信息被計算的方式。

統(tǒng)計信息怎么通過計劃被傳播的

一旦葉子節(jié)點的統(tǒng)計信息被計算出來，該統(tǒng)計信息會被傳播到其他節(jié)點。這里有兩種傳播方式:第一種（我們稱之為老方式）是非?；镜亩抑挥幸环N維度sizeInBytes被傳播，并且在各種操作中該維度被調(diào)整的的方式也是很基本的。例如，F(xiàn)ilter操作并不調(diào)整sizeInBytes的值，如下所示：

(
  spark.table(table_name)
  .filter(col("user_id") < 0)
).explain(mode="cost")

在這個查詢中，我們過濾除了所有user_id是負數(shù)的記錄，實際上是沒有該記錄的，但是spark并不能獲取這種信息，因為這種需要列級別的統(tǒng)計信息，這再老方式中不會被使用到。所以從這查詢計劃中可以看到，只有sizeInBytes被傳播，并且在兩個操作中該值保持不變.

第二種統(tǒng)計信息傳播的方式更加成熟，從spark 2.2開始但是它要求開啟CBO,而且要求通過ATC讓元數(shù)據(jù)儲存統(tǒng)計信息。這個時候所有的信息都會被傳播，加入我們提供了列級別的維度，spark將會將會計算filter操作，并且計算出一個更好的統(tǒng)計信息： 如你所見，在fiter操作的統(tǒng)計信息被改變了，rowCount非零，sizeInBytes 是1B，是最小值，從這個user_id列級別的統(tǒng)計信息，spark能夠知道負user_id的記錄是存在的，這個在查詢計劃中可以反映出來。
在這種新方式中，為了計算sizeInBytes，spark首先根據(jù)每個數(shù)據(jù)類型計算出單行的大小，之后再乘以rowCount去得到最終的sizeInBytes。假如rowCount是零，則sizeInBytes將會設置為1去避免在其他統(tǒng)計算的除零問題。這也適用于project操作（spark知道哪些列將會被投影，所以需要提前計算出單行的大小）

統(tǒng)計信息怎么被使用

此時我們已經(jīng)知道了統(tǒng)計信息怎么被計算的以及怎么通過邏輯計劃傳播的，現(xiàn)在讓我們來看一下在查詢計劃中怎么被使用以獲取更優(yōu)的計劃。
這有兩個地方統(tǒng)計信息會被使用：第一個是JoinSelection策略，這里spark將會決定使用哪種算法進行join兩個DataFrame（更多的細節(jié)參考 這里。基本的邏輯就是假如一個df小于某個閾值，spark將會使用BraodcastHashJoin（BHJ）,因為假如被廣播的df如果很小的話，這將是一個非常有效的方式。這個閾值通過spark.sql.autoBroadcastJoinThreshold 配置，默認是10MB，所以對于df的大小有個很好的預估的話，能夠幫助我們選擇一個更好的join優(yōu)化短發(fā)。
第二個地方也是跟join相關，即joinRecorder規(guī)則，使用這個規(guī)則 spark將會找到join操作最優(yōu)化的順序(如果你join多于兩個表的話)。這個規(guī)則默認是關閉的，假如你想開啟的話，通過如下配置：

spark.conf.set("spark.sql.cbo.joinReorder.enabled",True)

我們可以通過一下屬性控制df的最大數(shù)量：

spark.conf.set("spark.sql.cbo.joinReorder.dp.threshold",n)

n的默認值是12。

什么時候使用 ANALYZE TABLE command（ATC）?

我們已經(jīng)知道假如一個表是分區(qū)的，并且我們沒有運行ATC，spark將會使用默認的值 8EIB，這是很大的。所以在我們join很多表并且這些表是分區(qū)且十分小的情況下，他們是可以進行BHJ的，并且運行ATC是有意義的。當然我們必須清楚，加入一個表的數(shù)據(jù)被追加或者被覆蓋了,之前的統(tǒng)計信息就會被刪除，所以我們必須重新運行ATC。在某些情況下，更新元數(shù)據(jù)的統(tǒng)計信息是比較復雜的。一個解決方法是利用自適應查詢-spark 3.0的新特性。

自適應查詢

在spark 3.0 自適應查詢（AQE）這個新特性被開發(fā)，它將會以一種更加高級的方式使用統(tǒng)計信息。假如開啟了AQE（默認不開啟）,在每個stage執(zhí)行完后，統(tǒng)計信息會被重新計算。這就可以獲取更加精確的統(tǒng)計信息，以便能夠決定是否使用BHJ，AQE自身是一個很大的主題，我們分幾篇文章來介紹它。

看完上述內(nèi)容，你們掌握如何根據(jù)Spark SQL explaind中的統(tǒng)計信息深入了解CBO優(yōu)化的方法了嗎？如果還想學到更多技能或想了解更多相關內(nèi)容，歡迎關注億速云行業(yè)資訊頻道，感謝各位的閱讀！