『壹』 Spark 數據傾斜及其解決方案
本文從數據傾斜的危害、現象、原因等方面，由淺入深闡述Spark數據傾斜及其解決方案。
一、什麼是數據傾斜
對 Spark/Hadoop 這樣的分布式大數據系統來講，數據量大並不可怕，可怕的是數據傾斜。
對於分布式系統而言，理想情況下，隨著系統規模（節點數量）的增加，應用整體耗時線性下降。如果一台機器處理一批大量數據需要120分鍾，當機器數量增加到3台時，理想的耗時為120 / 3 = 40分鍾。但是，想做到分布式情況下每台機器執行時間是單機時的1 / N，就必須保證每台機器的任務量相等。不幸的是，很多時候，任務的分配是不均勻的，甚至不均勻到大部分任務被分配到個別機器上，其它大部分機器所分配的任務量只佔總得的小部分。比如一台機器負責處理 80% 的任務，另外兩台機器各處理 10% 的任務。
『不患多而患不均』，這是分布式環境下最大的問題。意味著計算能力不是線性擴展的，而是存在短板效應: 一個 Stage 所耗費的時間，是由最慢的那個 Task 決定。
由於同一個 Stage 內的所有 task 執行相同的計算，在排除不同計算節點計算能力差異的前提下，不同 task 之間耗時的差異主要由該 task 所處理的數據量決定。所以，要想發揮分布式系統並行計算的優勢，就必須解決數據傾斜問題。
二、數據傾斜的危害
當出現數據傾斜時，小量任務耗時遠高於其它任務，從而使得整體耗時過大，未能充分發揮分布式系統的並行計算優勢。
另外，當發生數據傾斜時，部分任務處理的數據量過大，可能造成內存不足使得任務失敗，並進而引進整個應用失敗。
三、數據傾斜的現象
當發現如下現象時，十有八九是發生數據傾斜了:
四、數據傾斜的原因
在進行 shuffle 的時候，必須將各個節點上相同的 key 拉取到某個節點上的一個 task 來進行處理，比如按照 key 進行聚合或 join 等操作。此時如果某個 key 對應的數據量特別大的話，就會發生數據傾斜。比如大部分 key 對應10條數據，但是個別 key 卻對應了100萬條數據，那麼大部分 task 可能就只會分配到10條數據，然後1秒鍾就運行完了；但是個別 task 可能分配到了100萬數據，要運行一兩個小時。
因此出現數據傾斜的時候，Spark 作業看起來會運行得非常緩慢，甚至可能因為某個 task 處理的數據量過大導致內存溢出。
五、問題發現與定位
1、通過 Spark Web UI
通過 Spark Web UI 來查看當前運行的 stage 各個 task 分配的數據量（Shuffle Read Size/Records），從而進一步確定是不是 task 分配的數據不均勻導致了數據傾斜。
知道數據傾斜發生在哪一個 stage 之後，接著我們就需要根據 stage 劃分原理，推算出來發生傾斜的那個 stage 對應代碼中的哪一部分，這部分代碼中肯定會有一個 shuffle 類運算元。可以通過 countByKey 查看各個 key 的分布。
2、通過 key 統計
也可以通過抽樣統計 key 的出現次數驗證。
由於數據量巨大，可以採用抽樣的方式，對數據進行抽樣，統計出現的次數，根據出現次數大小排序取出前幾個:
如果發現多數數據分布都較為平均，而個別數據比其他數據大上若干個數量級，則說明發生了數據傾斜。
六、如何緩解數據傾斜
基本思路
思路1. 過濾異常數據
如果導致數據傾斜的 key 是異常數據，那麼簡單的過濾掉就可以了。
首先要對 key 進行分析，判斷是哪些 key 造成數據傾斜。具體方法上面已經介紹過了，這里不贅述。
然後對這些 key 對應的記錄進行分析:
解決方案
對於第 1，2 種情況，直接對數據進行過濾即可。
第3種情況則需要特殊的處理，具體我們下面詳細介紹。
思路2. 提高 shuffle 並行度
Spark 在做 Shuffle 時，默認使用 HashPartitioner（非 Hash Shuffle）對數據進行分區。如果並行度設置的不合適，可能造成大量不相同的 Key 對應的數據被分配到了同一個 Task 上，造成該 Task 所處理的數據遠大於其它 Task，從而造成數據傾斜。
如果調整 Shuffle 時的並行度，使得原本被分配到同一 Task 的不同 Key 發配到不同 Task 上處理，則可降低原 Task 所需處理的數據量，從而緩解數據傾斜問題造成的短板效應。
（1）操作流程
RDD 操作 可在需要 Shuffle 的操作運算元上直接設置並行度或者使用 spark.default.parallelism 設置。如果是 Spark sql，還可通過 SET spark.sql.shuffle.partitions=[num_tasks] 設置並行度。默認參數由不同的 Cluster Manager 控制。
dataFrame 和 sparkSql 可以設置 spark.sql.shuffle.partitions=[num_tasks] 參數控制 shuffle 的並發度，默認為200。
（2）適用場景
大量不同的 Key 被分配到了相同的 Task 造成該 Task 數據量過大。
（3）解決方案
調整並行度。一般是增大並行度，但有時如減小並行度也可達到效果。
（4）優勢
實現簡單，只需要參數調優。可用最小的代價解決問題。一般如果出現數據傾斜，都可以通過這種方法先試驗幾次，如果問題未解決，再嘗試其它方法。
（5）劣勢
適用場景少，只是讓每個 task 執行更少的不同的key。無法解決個別key特別大的情況造成的傾斜，如果某些 key 的大小非常大，即使一個 task 單獨執行它，也會受到數據傾斜的困擾。並且該方法一般只能緩解數據傾斜，沒有徹底消除問題。從實踐經驗來看，其效果一般。
思路3. 自定義 Partitioner
（1）原理
使用自定義的 Partitioner（默認為 HashPartitioner），將原本被分配到同一個 Task 的不同 Key 分配到不同 Task。
例如，我們在 groupByKey 運算元上，使用自定義的 Partitioner:
（2）適用場景
大量不同的 Key 被分配到了相同的 Task 造成該 Task 數據量過大。
（3）解決方案
使用自定義的 Partitioner 實現類代替默認的 HashPartitioner，盡量將所有不同的 Key 均勻分配到不同的 Task 中。
（4）優勢
不影響原有的並行度設計。如果改變並行度，後續 Stage 的並行度也會默認改變，可能會影響後續 Stage。
（5）劣勢
適用場景有限，只能將不同 Key 分散開，對於同一 Key 對應數據集非常大的場景不適用。效果與調整並行度類似，只能緩解數據傾斜而不能完全消除數據傾斜。而且需要根據數據特點自定義專用的 Partitioner，不夠靈活。
思路4. Rece 端 Join 轉化為 Map 端 Join
通過 Spark 的 Broadcast 機制，將 Rece 端 Join 轉化為 Map 端 Join，這意味著 Spark 現在不需要跨節點做 shuffle 而是直接通過本地文件進行 join，從而完全消除 Shuffle 帶來的數據傾斜。
其中 A 是比較小的 dataframe 並且能夠整個存放在 executor 內存中。
（1）適用場景
參與Join的一邊數據集足夠小，可被載入進 Driver 並通過 Broadcast 方法廣播到各個 Executor 中。
（2）解決方案
在 Java/Scala 代碼中將小數據集數據拉取到 Driver，然後通過 Broadcast 方案將小數據集的數據廣播到各 Executor。或者在使用 SQL 前，將 Broadcast 的閾值調整得足夠大，從而使 Broadcast 生效。進而將 Rece Join 替換為 Map Join。
（3）優勢
避免了 Shuffle，徹底消除了數據傾斜產生的條件，可極大提升性能。
（4）劣勢
因為是先將小數據通過 Broadcase 發送到每個 executor 上，所以需要參與 Join 的一方數據集足夠小，並且主要適用於 Join 的場景，不適合聚合的場景，適用條件有限。
思路5. 拆分 join 再 union
思路很簡單，就是將一個 join 拆分成 傾斜數據集 Join 和 非傾斜數據集 Join，最後進行 union:
（1）適用場景
兩張表都比較大，無法使用 Map 端 Join。其中一個 RDD 有少數幾個 Key 的數據量過大，另外一個 RDD 的 Key 分布較為均勻。
（2）解決方案
將有數據傾斜的 RDD 中傾斜 Key 對應的數據集單獨抽取出來加上隨機前綴，另外一個 RDD 每條數據分別與隨機前綴結合形成新的RDD（相當於將其數據增到到原來的N倍，N即為隨機前綴的總個數），然後將二者Join並去掉前綴。然後將不包含傾斜Key的剩餘數據進行Join。最後將兩次Join的結果集通過union合並，即可得到全部Join結果。
（3）優勢
相對於 Map 則 Join，更能適應大數據集的 Join。如果資源充足，傾斜部分數據集與非傾斜部分數據集可並行進行，效率提升明顯。且只針對傾斜部分的數據做數據擴展，增加的資源消耗有限。
（4）劣勢
如果傾斜 Key 非常多，則另一側數據膨脹非常大，此方案不適用。而且此時對傾斜 Key 與非傾斜 Key 分開處理，需要掃描數據集兩遍，增加了開銷。
思路6. 大表 key 加鹽，小表擴大 N 倍 jion
如果出現數據傾斜的 Key 比較多，上一種方法將這些大量的傾斜 Key 分拆出來，意義不大。此時更適合直接對存在數據傾斜的數據集全部加上隨機前綴，然後對另外一個不存在嚴重數據傾斜的數據集整體與隨機前綴集作笛卡爾乘積（即將數據量擴大N倍）。
其實就是上一個方法的特例或者簡化。少了拆分，也就沒有 union。
（1）適用場景
一個數據集存在的傾斜 Key 比較多，另外一個數據集數據分布比較均勻。
（2）優勢
對大部分場景都適用，效果不錯。
（3）劣勢
需要將一個數據集整體擴大 N 倍，會增加資源消耗。
思路7. map 端先局部聚合
在 map 端加個 combiner 函數進行局部聚合。加上 combiner 相當於提前進行 rece ,就會把一個 mapper 中的相同 key 進行聚合，減少 shuffle 過程中數據量 以及 rece 端的計算量。這種方法可以有效的緩解數據傾斜問題，但是如果導致數據傾斜的 key 大量分布在不同的 mapper 的時候，這種方法就不是很有效了。
思路8. 加鹽局部聚合 + 去鹽全局聚合
這個方案的核心實現思路就是進行兩階段聚合。第一次是局部聚合，先給每個 key 都打上一個 1~n 的隨機數，比如 3 以內的隨機數，此時原先一樣的 key 就變成不一樣的了，比如 (hello, 1) (hello, 1) (hello, 1) (hello, 1) (hello, 1)，就會變成 (1_hello, 1) (3_hello, 1) (2_hello, 1) (1_hello, 1) (2_hello, 1)。接著對打上隨機數後的數據，執行 receByKey 等聚合操作，進行局部聚合，那麼局部聚合結果，就會變成了 (1_hello, 2) (2_hello, 2) (3_hello, 1)。然後將各個 key 的前綴給去掉，就會變成 (hello, 2) (hello, 2) (hello, 1)，再次進行全局聚合操作，就可以得到最終結果了，比如 (hello, 5)。
不過進行兩次 maprece，性能稍微比一次的差些。
七、Hadoop 中的數據傾斜
Hadoop 中直接貼近用戶使用的是 Maprece 程序和 Hive 程序，雖說 Hive 最後也是用 MR 來執行（至少目前 Hive 內存計算並不普及），但是畢竟寫的內容邏輯區別很大，一個是程序，一個是Sql，因此這里稍作區分。
Hadoop 中的數據傾斜主要表現在 ruce 階段卡在99.99%，一直99.99%不能結束。
這里如果詳細的看日誌或者和監控界面的話會發現:
經驗: Hive的數據傾斜，一般都發生在 Sql 中 Group 和 On 上，而且和數據邏輯綁定比較深。
優化方法
這里列出來一些方法和思路，具體的參數和用法在官網看就行了。
說明
八、參考文章
『貳』 數據傾斜（一）：數據傾斜及具體場景
相信大部分做數據的童鞋們都會遇到數據傾斜，數據傾斜會發生在數據開發的各個環節中，比如：
1.用Hive算數據的時候rece階段卡在99.99%
2.用SparkStreaming做實時演算法時候，一直會有executor出現OOM的錯誤，但是其餘的executor內存使用率卻很低。
3.這些問題經常會困擾我們，辛辛苦苦等了幾個小時的數據就是跑不出來，心裡多難過啊。
為什麼要突出這么大數據量？先說一下筆者自己最初對數據量的理解：
數據量大就了不起了？數據量少，機器也少，計算能力也是有限的，因此難度也是一樣的。憑什麼數據量大就會有數據傾斜，數據量小就沒有？
這樣理解也有道理，但是比較片面，舉兩個場景來對比：
公司一：總用戶量1000萬，5台64G內存的的伺服器。
公司二：總用戶量10億，1000台64G內存的伺服器。
兩個公司都部署了Hadoop集群。假設現在遇到了數據傾斜，發生什麼？
1.公司一的數據分析師在做join的時候發生了數據傾斜，會導致有幾百萬用戶的相關數據集中到了一台伺服器上，幾百萬的用戶數據，說大也不大，正常欄位量的數據的話64G還是能輕松處理掉的。
2.公司二的數據分析師在做join的時候也發生了數據傾斜，可能會有1個億的用戶相關數據集中到了一台機器上了（相信我，這很常見）。這時候一台機器就很難搞定了，最後會很難算出結果。
下面會分幾個場景來描述一下數據傾斜的特徵，方便讀者辨別。由於Hadoop和Spark是最常見的兩個計算平台，下面就以這兩個平台說明。
Hadoop中直接貼近用戶使用使用的時Maprece程序和Hive程序，雖說Hive最後也是用MR來執行（至少目前Hive內存計算並不普及），但是畢竟寫的內容邏輯區別很大，一個是程序，一個是Sql，因此這里稍作區分。
具體表現：
Hadoop中的數據傾斜主要表現在： Rece階段卡在99.99%，一直不能結束。
這里如果詳細的看日誌或者和監控界面的話會發現：
Spark中的數據傾斜也很常見，這里包括Spark Streaming和Spark Sql，表現主要有下面幾種：
『叄』 hive運行sql rece 為1 ，跑不動怎麼處理
1.jpg 優化可以從幾個方面著手：1. 好的模型設計事半功倍。2. 解決數據傾斜問題。3. 減少job數。4. 設置合理的map rece的task數，能有效提升性能。(比如，10w+級別的計算，用160個rece，那是相當的浪費，1個足夠)。5. 自己動手寫sql解決數據傾斜問題是個不錯的選擇。set hive.groupby.skewindata=true;這是通用的演算法優化，但演算法優化總是漠視業務，習慣性提供通用的解決方法。 Etl開發人員更了解業務，更了解數據，所以通過業務邏輯解決傾斜的方法往往更精確，更有效。6. 對count(distinct)採取漠視的方法，尤其數據大的時候很容易產生傾斜問題，不抱僥幸心理。自己動手，豐衣足食。7. 對小文件進行合並，是行至有效的提高調度效率的方法，假如我們的作業設置合理的文件數，對雲梯的整體調度效率也會產生積極的影響。8. 優化時把握整體，單個作業最優不如整體最優。
『肆』 hive數據傾斜及處理
火山日常啰嗦
學習了一些大數據的相關框架後，發現應用層的東西確實不難，真正難的都是底層原理，所以我查看了很多資料，借鑒了前人的方法再加上自己的理解，寫下了這篇文章。
數據傾斜的直白概念：
數據傾斜就是數據的分布不平衡，某些地方特別多，某些地方又特別少，導致的在處理數據的時候，有些很快就處理完了，而有些又遲遲未能處理完，導致整體任務最終遲遲無法完成，這種現象就是數據傾斜。
針對maprece的過程來說就是，有多個rece，其中有一個或者若干個rece要處理的數據量特別大，而其他的rece處理的數據量則比較小，那麼這些數據量小的rece很快就可以完成，而數據量大的則需要很多時間，導致整個任務一直在等它而遲遲無法完成。
跑mr任務時常見的rece的進度總是卡在99%，這種現象很大可能就是數據傾斜造成的。
產生數據傾斜的原因：
1） key的分布不均勻或者說某些key太集中。
上面就說過，rece的數據量大小差異過大，而rece的數據是分區的結果，分區是對key求hash值，根據hash值決定該key被分到某個分區，進而進入到某個rece，而如果key很集中或者相同，那麼計算得到它們的hash值可能一樣，那麼就會被分配到同一個rece，就會造成這個rece所要處理的數據量過大。
2） 業務數據自身的特性。
比如某些業務數據作為key的欄位本就很集中，那麼結果肯定會導致數據傾斜啊。
還有其他的一些原因，但是，根本原因還是key的分布不均勻，而其他的原因就是會造成key不均勻，進而導致數據傾斜的後果，所以說根本原因是key的分布不均勻。
既然有數據傾斜這種現象，就必須要有數據傾斜對應的處理方案啊。
簡單地說數據傾斜這種現象導致的任務遲遲不能完成，耗費了太多時間，極大地影響了性能，所以我們數據傾斜的解決方案設計思路就是往如何提高性能，即如何縮短任務的處理時間這方面考慮的，而要提高性能，就要讓key分布相對均衡，所以我們的終極目標就是考慮如何預處理數據才能夠使得它的key分布均勻。
常見的數據傾斜處理方案：
1 設置參數
1）設置hive.map.aggr=true //開啟map端部分聚合功能，就是將key相同的歸到一起，減少數據量，這樣就可以相對地減少進入rece的數據量，在一定程度上可以提高性能，當然，如果數據的減少量微乎其微，那對性能的影響幾乎沒啥變化。
2）設置hive.groupby.skewindata=true //如果發生了數據傾斜就可以通過它來進行負載均衡。當選項設定為 true，生成的查詢計劃會有兩個 MR Job。第一個 MR Job 中，Map 的輸出結果集合會隨機分布到 Rece 中，每個 Rece 做部分聚合操作，並輸出結果，這樣處理的結果是相同的Key 有可能被分發到不同的 Rece 中，從而達到負載均衡的目的；第二個 MR Job 再根據預處理的數據結果按照Key 分布到 Rece 中（這個過程是按照key的hash值進行分區的，不同於mr job1的隨機分配，這次可以保證相同的Key 被分布到同一個 Rece 中），最後完成最終的聚合操作。所以它主要就是先通過第一個mr job將key隨機分配到rece，使得會造成數據傾斜的key可能被分配到不同的rece上，從而達到負載均衡的目的。到第二個mr job中，因為第一個mr job已經在rece中對這些數據進行了部分聚合（就像單詞統計的例子，a這個字母在不同的rece中，已經算出它在每個rece中的個數，但是最終的總的個數還沒算出來，那麼就將它傳到第二個mr job，這樣就可以得到總的單詞個數），所以這里直接進行最後的聚合就可以了。
3）hive.exec.recers.bytes.per.recer=1000000000 （單位是位元組）
每個rece能夠處理的數據量大小，默認是1G
4）hive.exec.recers.max=999
最大可以開啟的rece個數，默認是999個
在只配了hive.exec.recers.bytes.per.recer以及hive.exec.recers.max的情況下，實際的rece個數會根據實際的數據總量/每個rece處理的數據量來決定。
5）mapred.rece.tasks=-1
實際運行的rece個數，默認是-1，可以認為指定，但是如果認為在此指定了，那麼就不會通過實際的總數據量/hive.exec.recers.bytes.per.recer來決定rece的個數了。
2 sql語句優化
給幾個具體的場景以及在這些場景下的處理方案：
1）進行表的join這種業務操作時，經常會產生數據傾斜。
原因就是這些業務數據本就存在key會分布不均勻的風險，所以我們join時不能使用普通的join（rece端join）或者可以使用普通join，但是是優化後的。
但是這種操作有個前提條件就是僅適用於小表join大表，而小表怎麼定義它的大小，多小的表才算小表，這里有個參數可以確定的（但是這個參數名我暫時忘記了），如果小表的數據大小小於這個值，就可以使用map join，而是在這種情況下是自動使用map join這種方案的。所以如果是大小表join，直接用map join，避免數據傾斜。
方法1：（普通join）
select * from log a join users b on （a.user_id is not null and a.user_id = b.user_id ）；
這是屬於表的內連接的，兩張表不滿足條件的記錄都不保留。
方法2：檢測到user_id是null時給它賦予一個新值（這個新值由一個字元串（比如我自己給它定一個 hive）加上一個隨機數組成），這樣就可以將原來集中的key分散開來，也避免了數據傾斜的風險。而且因為這些數據本來就是無效數據，根本不會出現在結果表中，所以，這樣處理user_id（由一個字元串（比如我自己給它定一個 hive）加上一個隨機數），它也無法關聯的，因為有效的數據的user_id沒有這種形式的，所以就算這些無效數據出現在不同的rece中還是不會影響結果的，我這樣處理只是為了將它們分散開而已，所以用這種方法處理，結果表中也不會出現null這些無效數據，跟過濾處理方案得到的結果是一樣的。（普通join）
select *
from log a
join users b
on case when a.user_id is null then concat(『hive』,rand() ) else a.user_id end = b.user_id;
但是這兩種方案只是適用於大表join大表的內連接，兩張表的無效數據都不保留。
但是如果對於左外連接或者右外連接這種情況，即使驅動表中某些記錄在另一張表中沒有數據與它對應，但我們是依然需要保留驅動表的這些數據的，那該怎麼辦呢？其實很簡單，只需要將上述方法得到的結果再與驅動表的這些無數據取並集就可以了。
如下：
select * from log a
left outer join users b
on a.user_id is not null
and a.user_id = b.user_id
union all
select * from log a
where a.user_id is null;
2)雖然都是大表，但是呢對於某些業務數據而言，其有用的部分只佔它所在表的很少一部分，那麼我們就可以將它們先取出來，得到的結果應該是一張小表，那麼就可以使用map join來避免數據傾斜了。
場景：用戶表中user_id欄位為int，log表中user_id欄位既有string類型也有int類型。
當按照user_id進行兩個表的Join操作時，因為我們在連接時要進行user_id的比較，所以需要user_id的類型都相同，如果我們選擇將log表中的String類型轉換為int類型，那麼就可能會出現這種情況：String類型轉換為int類型得到的都是null值（這就是類型轉換的問題了，String類型數據轉換為int類型會失敗，數據丟失，就會賦null值），如果所有的String類型的user_id都變成了null,那麼就又出現了集中的key，分區後就又會導致數據傾斜。所以我們進行類型轉換時不能選擇將String類型轉換為int，而應該將int類型轉換為String,因為int轉換為String不會出問題，int類型原來的值是什麼，轉換為String後對應的字元串就會是什麼，形式沒變，只是類型變了而已。
解決方法：把int類型轉換成字元串類型
select * from users a
join logs b
on (a.usr_id = cast(b.user_id as string));
比如有一份日誌，要你從日誌中統計某天有多少個用戶訪問網站，即統計有多少個不同的user_id；但是呢這個網站卻又恰巧遭到攻擊，日誌中大部分都是同一個user_id的記錄，其他的user_id屬於正常訪問，訪問量不會很大，在這種情況下，當你直接使用count(distinct user_id)時，這也是要跑mr任務的啊，這時這些大量的相同的user_id就是集中的key了，結果就是通過分區它們都被分到一個rece中，就會造成這個rece處理的數據特別大，而其中的rece處理的數據都很小，所以就會造成數據傾斜。
那麼要怎麼優化呢？
方法1：可以先找出這個user_id是什麼，過濾掉它，然後通過count(distinct user_id)計算出剩餘的那些user_id的個數，最後再加1（這1個就是那個被過濾掉的user_id，雖然它有大量的記錄，但是ser_id相同的都是同一個用戶，而我們要計算的就是用戶數）
sql語句展示：
分組求和後降序排序，就可以得到這個數據量最大的user_id是什麼，然後我們下一步操作時就過濾它，等計算完其他的再加上它這一個。
select user_id,count(user_id) from log group by user_id desc limit 2;
select count（distinct user_id）+1 as sum from log;
sum就是最終的結果--用戶數
方法2：我們可以先通過group by分組，然後再在分組得到的結果的基礎之上進行count
sql語句展示：
select count(*) from (select user_id from log group by user_id) new_log;
總的來說就是，數據傾斜的根源是key分布不均勻，所以應對方案要麼是從源頭解決（不讓數據分區，直接在map端搞定），要麼就是在分區時將這些集中卻無效的key過濾（清洗）掉，或者是想辦法將這些key打亂，讓它們進入到不同的rece中。
性能調優是指通過調整使得機器處理任務的速度更快，所花的時間更少，而數據傾斜的處理是hive性能調優的一部分，通過處理能夠大大減少任務的運行時間。
除了數據傾斜的處理之外，hive的優化還有其他方面的，例如where子句優化：
select * from a left outer join b on (a.key=b.key) where a.date='2017-07-11' and b.date='2017-07-11';
這是一個左外連接。
這個sql語句執行的結果是：得到的結果是表a與表b的連接表，且表中的記錄的date都是'2017-07-11'。
而這個sql語句的執行過程是：逐條獲取到a表的記錄，然後掃描b表，尋找欄位key值為a.key的記錄,找到後將b表的這條記錄連接到a表上，然後判斷連接後的這條記錄是否滿足條件a.date='2017-07-11' and b.date='2017-07-11'，如果滿足，則顯示，否則，丟棄。
因為這是一個左外連接，且a為驅動表，連接時在a中發現key而在b中沒有發現與之相等的key時，b中的列將置為null，包括列date，一個不為null，一個為null，這樣後邊的where就沒有用了。
簡答的說這個方案的做法就是先按連接條件進行連接，連接後再看where條件，如果不滿足就丟棄，那之前連接所做的那些功夫就浪費了，白白耗費了資源（cpu等），增加了運行的總時間，如果有一種方案可以在未進行連接之前就直接判斷出不滿足最終的條件，那麼就可以直接丟棄它，這樣對於這樣的記錄就不要浪費資源以及時間去連接了，這樣也是能提升性能的，下面就看看這種方案：
sql語句：
將剛才的where限制條件直接放到on裡面，那麼就變成了滿足這三個條件才會進行連接，不滿足的直接過濾掉，就像上面所說的，少了無效連接那一步，就相對地節約了時間，如果這樣的無效連接的記錄很多的話，那麼採用這種改進版的方案無疑能夠較大程度地提高性能。
select * from a left outer join b on (a.key=b.key and a.date='2017-07-11' and b.date='2017-07-11'）;
不管怎麼說，我們在運行任務時，總是希望能加快運行速度，縮短運行時間，更快地得到結果，即提升性能，這是我們的目的，這就是我們所謂的性能調優。
關於小表join大表的補充：
表join時的操作是這樣的：
當操作到驅動表的某條記錄時，就會全局掃描另一張表，尋找滿足條件的記錄，而當掃描它時，為了讀取速度更快，一般都選先將它載入到內存，而內存的大小是有限的，為了不佔據過多的內存或者避免內存溢出，載入進入內存的表一般是小表，即數據量比較小，map join就是這樣做的。
即驅動表不放進內存，而另一張表（即要連接到驅動表的那一張表）就要先載入進內存，為了掃描速度更快，提高性能。
比如select * from a left outer join b on (a.key=b.key);
左外連接，驅動表是a，表b的記錄是要被連接到表a的，所以每在a上連接一條記錄就要被全局掃描一次的表是b，所以表b應先載入到內存（前提是表b是小表，如果是大表的話，估計會產生oom異常--out of memory內存溢出異常）。
select * from aa right outer join bb on (a.key=b.key);
右外連接，驅動表是bb，aa應先載入到內存（前提是小表）。
ps：希望我的分享能幫助到有需要的夥伴哦。我不是大神的哦，如果文中有誤，還請大家不吝賜教，幫忙指正，謝謝了！！！