valresult=txDF.select("price","volume",.join(users,Seq("userId"),.groupBy(col("name"),col("age")).agg(sum(col("price")*兩表關(guān)聯(lián)的查詢語句經(jīng)過轉(zhuǎn)換之后，得到的OptimizedLogicaln如下圖所示。注意，在邏輯計劃的根節(jié)點，出現(xiàn)了“JoinInner”字樣，Catalyst優(yōu)化器明確了這一步需要做內(nèi)關(guān)聯(lián)。但是，怎么做內(nèi)關(guān)聯(lián)，使用哪種Join策略來進行關(guān)聯(lián)，Catalyst并沒有交代清楚。小Q變身：Optimized 為了讓查詢計劃（Queryn）變得可操作、可執(zhí)行，Catalyst的物理優(yōu)化階段（Physicalnning）可以分為兩個環(huán)節(jié)：優(yōu)化Sparkn和生成Physicaln在優(yōu)化Spark n的過程中，Catalyst基于既定的優(yōu)化策略（Strategies），把邏輯計劃中的關(guān)系操作符一一映射成物理操作符，生成Spark 在生成Physical n過程中，Catalyst再基于事先定義的PreparationRules，對 n做進一步的完善、生成可執(zhí)行的Physical 那么問題來了，在優(yōu)化rk nCatalyt數(shù)量上來說，Catalyst有4類優(yōu)化略，其中有6類和流計算有關(guān)，剩下的8類適用于所有的計算場景，如批處理、數(shù)據(jù)分析、機器學(xué)習(xí)和圖計算，當(dāng)然也包括流計算。因此，我們只需了解這8類優(yōu)化略。Functions），把邏輯計劃中的操作符平行映射為Spark n中的物理算子。比如，BasicOperators策略直接把Project、Filter、Sort等邏輯操作符平行地映射為物理操作符。其他策略的優(yōu)化過程也類似，因此，在優(yōu)化Spark n這一環(huán)節(jié)，咱們只要抓住一遴選上面的8Jointn。接下來，我們就以JoinSelectn為例，詳細講解這一環(huán)節(jié)的優(yōu)化過程。JoinSelection優(yōu)化階段采用哪種Join策略。那么問題來了，Catalyst都有哪些Join策略？Catalyst都有哪些Join結(jié)合Joins的實現(xiàn)機制和數(shù)據(jù)的分發(fā)方式，Catalyst在運行時總共支持5種Join策略，分別是BroadcastHashJoin（BHJ）、ShuffleSortMergeJoin（SMJ）、ShuffleHashJoin（SHJ）、BroadcastNestedLoopJoin（BNLJ）ShuffleCartesianProductJoin通過上表中5種Join策略的含義，我們知道，它們是來自2種數(shù)據(jù)分發(fā)方式（廣播和Shuffle）與3種Join實現(xiàn)機制（HashJoins、SortMergeJoins和NestedLoopJoins）的排列組合。那么，在JoinSelection的優(yōu)化過程中，Catalyst會基于什么邏輯，優(yōu)先選擇哪種Join策略呢？JoinSelectionJoin邏輯其實很簡單，Catalyst總會嘗試優(yōu)先選擇執(zhí)行效率最高的策略。具體來說，在選擇join策略的時候，JoinSelection會先判斷當(dāng)前查詢是否滿足BHJ所要求的先決條件：如果滿足就立即選中BHJ；如果不滿足，就繼續(xù)判斷當(dāng)前查詢是否滿足SMJ的先決條件。以此類推，直到最終選無可選，用CPJ來兜底。5JoinJoinSelection總的來說，這些信息分為兩大類，第一類是“條件型”信息，用來5大Join策略的先決條件。第二類是“指令型”信息，也就是開發(fā)者提供的JoinHints。我們先來說條件型信息，它包含兩種。第一種是Join形式等，這種信息的來源是查詢語句本身。第二種是內(nèi)表尺寸，這些信息的來源就比較廣泛了，可以是e表之上的 YEBE語句，可以是k于Parquet、ORCCSV等源文的尺寸預(yù)估，甚至是來自AQE的動態(tài)統(tǒng)計信息。大指令型信息也就是JoinHints，它的種類非常豐富，它允許我們把個人意志凌駕于SparkSQL之上。比如說，如果我們對小Q的查詢語句做了如下的調(diào)整，JoinSelection在做JoinSMJSHJ。具體的valresult=txDF.select("price","volume",.join(users.hint("shuffle_hash"),Seq("userId"),.groupBy(col("name"),col("age")).agg(sum(col("price")JointnJoinQ是怎么選擇的。小QJoin策略的時候，Jointn優(yōu)先嘗試判斷小QBJ顯然，小Q是等值的InnerJoin，因此表格中BHJ那一行的前兩個條件小Q都滿足。但是，內(nèi)表users尺寸較大，超出了廣播閾值的默認值10MB，不滿足BHJ的第三個條件。因此，JoinSelection不得不、放棄BHJ策略，只好退而求其次，沿著表格繼續(xù)向下，嘗試判斷小Q是否滿足SMJ的先決條件。SMJ的先決條件很寬松，查詢語句只要是等值Join就可以。小Q自然是滿足這個條件的，因此JoinSelection最終給小Q選定的Join策略就是SMJ。下圖是小Q優(yōu)化過后的 現(xiàn)在我們知道了Catalyst都有哪些Join策略，JoinSelection如何對不同的Join策略做選擇。小Q也從OptimizedLogicaln搖身一變，轉(zhuǎn)換成了Sparkn，也明確了在運行時采用SMJ來做關(guān)聯(lián)計算。不過，即使小Q在Sparkn中已經(jīng)明確了每一步該“怎么做”，但是，Spark還是做不到把這樣的查詢計劃轉(zhuǎn)化成可執(zhí)行的分布式任務(wù)，生成 原來，ShuffleSortMergeJoin的計算需要兩個先決條件：Shuffle和排序。而Sparkn中并沒有明確指定以哪個字段為基準進行Shuffle，以及按照哪個字段去做排序。因此，Catalyst需要對Spark n做進一步的轉(zhuǎn)換，生成可操作、可執(zhí)行的Physicaln。具體怎么做呢？我們結(jié)合Catalyst物理優(yōu)化階段的流程圖來詳細講講。從上圖中我們可以看到，從Spark n到Physical n的轉(zhuǎn)換，需要幾組叫做PreparationRules的規(guī)則。這些規(guī)則堅守最后一班崗，負責(zé)生成Physical PreparationPreparationRules有6組規(guī)則，這些規(guī)則作用到Spark n之上就會生成Physicaln，而Physical n最終會由Tungsten轉(zhuǎn)化為用于計算RDD的分布式任務(wù)。QPythonUDF。因此，在小Q的例子中，我們并不會用到子查詢、數(shù)據(jù)復(fù)用或是PythonUDF之類的規(guī)則，只有EnsureRequirements和CollapseCodegenStages這兩組規(guī)則會用到小Q的PhysicalnEnsureRequirementsCollapseCodegenStages上就是Tungsten的WSCG功能，我們下一講再詳細說。EnsureRequirements計劃中的每一個操作符節(jié)點，都有4個屬性用來分別描述數(shù)據(jù)輸入和輸出的分布狀態(tài)。EnsureRequirements我們以小Q的Spark n樹形結(jié)構(gòu)圖為例，可以看到：圖中左右兩個分支分別表示掃描和處理users表和transactions表。在樹的最頂端，根節(jié)點SortMergeJoin有兩個Project子節(jié)點，它們分別用來表示users表和transactions表上的投影數(shù)據(jù)。這兩個Project的outputPartitioning屬性和outputOrdering屬性分別是Unknow和None。因此，它們輸出的數(shù)據(jù)沒有按照任何列進行Shuffle或是排序。但是，SortMergeJoin對于輸入數(shù)據(jù)的要求很明確：按照userId分成200個分區(qū)且排好序，而這兩個Project子節(jié)點的輸出顯然并沒有滿足父節(jié)點SortMergeJoin的要求。這個時候，EnsureRequirements規(guī)則就要介入了，它通過添加必要的操作符，如Shuffle和排序，來保證SortMergeJoin節(jié)點對于輸入數(shù)據(jù)的要求一定要得到滿足，如下圖所在兩個Project節(jié)點之后，EnsureRequirements規(guī)則分別添加了Exchange和Sort節(jié)ExchangeShuffleSortMergeJoin求；Sort表示排序，用于滿足SortMergeJoin對于數(shù)據(jù)有序的要求。添加了必需的節(jié)點之后，小Q的Physical n已經(jīng)相當(dāng)具體了。這個時候，Spark可以通過調(diào)用Physical n的doExecute方法，把結(jié)構(gòu)化查詢的計算結(jié)果，轉(zhuǎn)換成RDD[InternalRow]，這里的InternalRow，就是Tungsten設(shè)計的定制化二進制數(shù)據(jù)結(jié)RDD[InternalRow]之上的Action算子，Spark就可以觸發(fā)Physical n從頭至尾依序最后，我們再來看看小Q 首先，我們看到EnsureRequirements規(guī)則在兩個分支的頂端分別添加了Exchange和Sort操作，來滿足根節(jié)點SortMergeJoin的計算需要。其次，如果你仔細觀察的話，會發(fā)現(xiàn)Physical n中多了很多星號“*”，這些星號的后面還帶著括號和數(shù)字，如圖中的“*（3）”、“*（1）”。這種星號“*”標記表示的就是WSCG，后面的數(shù)字代表Stage編號。因此，括號中數(shù)字相同的操，最終都會被捏合成一份“手寫代碼”，也就是我們下一講要說的Tungsten的WSCG。Q士”，Catalyst這位人生導(dǎo)師在其中的作用功不可沒。為了把邏輯計劃轉(zhuǎn)換為可以交付執(zhí)行的物理計劃，SparkSQL物理優(yōu)化階段包含兩個環(huán)節(jié)：優(yōu)化Spark n和生成Physical 在優(yōu)化Sparkn這個環(huán)節(jié)，Catalyst基于既定的策略把邏輯計劃平行映射為n。策略很多，我們重點掌握JoinSelection策略就可以，它被用來在運行時選擇最佳的JoinJoinSelectionBHJSMJSHJBNLJCPJ句是否滿足每一種Join策略的先決條件進行“擇優(yōu)錄取”。JoinSelectionBHJ>SMJ>SHJ>>CPJ，還可以通過在SQL或是DSLJoinhints，來明確地指定Join在生成Physical n這個環(huán)節(jié)，Catalyst基于既定的幾組PreparationRules，把優(yōu)化過后的Spark n轉(zhuǎn)換成可以交付執(zhí)行的物理計劃，也就是Physical PreparationRules當(dāng)中，你需要重點掌握EnsureRequirements規(guī)則。EnsureRequirements用來確保每一個操作符的輸入條件都能夠得到滿足，在必要的時候，會把必需的操作行插入到Physical n中。比如對于ShuffleSortMergeJoin和3Join26JoinCatalyst有支持BroadcastSortMergeJoin策略？ 售賣。頁面已增加防盜追蹤，將依 上一 21|Catalyst邏輯計劃：你的SQL語句是怎么被優(yōu)化的？（上下一 23|鎢絲計劃：Tungsten給開發(fā)者帶來了哪些福報個人認為沒有必要：因為被廣去的數(shù)據(jù)集合都很小，可以通過Hash或者NestedLoop來實現(xiàn)，而這二者之間的區(qū)別在于HashNestedLoop作者回復(fù):標準答案了，滿分但如果細想的話，你會發(fā)現(xiàn)，對于非等值Jn來說，即便你打算嘗試用SM去實現(xiàn)，最終實現(xiàn)下來你發(fā)現(xiàn)算法的復(fù)雜度，和你使用NesdpJn雙層for循環(huán)的算法是等價的。而SM外的排序開銷，因此里外里算下來，還不如直接用NL也就是來實現(xiàn)。1 2作者回復(fù):正解，滿分??而Broadcast本身就是對數(shù)據(jù)的全量發(fā)送，并不需要區(qū)分數(shù)據(jù)歸屬于哪個partition。提問：老師說Tungsten的WSCG會將同個stage內(nèi)的操作捏合成一個手寫代碼，意思kingcall同學(xué)的答案哈后面的提問是個好問題。邏輯優(yōu)化、物理優(yōu)化、n優(yōu)化，是rkQL一套”，這3部分就是在講，一份代碼（DataFrm、Datase、L）GSh