‘壹’ 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：希望我的分享能帮助到有需要的伙伴哦。我不是大神的哦，如果文中有误，还请大家不吝赐教，帮忙指正，谢谢了！！！