1 Spark-Join

参数	默认值	说明
spark.sql.shuffle.partitions	200	Configures the number of partitions to use when shuffling data for joins or aggregations.
spark.default.parallelism	For distributed shuffle operations like reduceByKey and join, the largest number of partitions in a parent RDD. For operations like parallelize with no parent RDDs, it depends on the cluster manager:Local mode: number of cores on the local machineMesos fine grained mode: 8 Others: total number of cores on all executor nodes or 2, whichever is larger	Default number of partitions in RDDs returned by transformations like join, reduceByKey, and parallelize when not set by user.

上面两个参数都是设置默认的并行度，但是适用的场景不同：

spark.sql.shuffle.partitions是对sparkSQL进行shuffle操作的时候生效，比如 join或者aggregation等操作的时候，之前有个同学设置了spark.default.parallelism 这个并行度为2000，结果还是产生200的stage，排查了很久才发现，是这个原因。
spark.default.parallelism这个参数只是针对rdd的shuffle操作才生效，比如join，reduceByKey。

作者：pcqlegend
链接：https://www.jianshu.com/p/c5914126ef98
来源：简书
著作权归作者所有。商业转载请联系作者获得授权，非商业转载请注明出处。

1.1 shuffle join VS map join

在对大表与大表之间进行连接操作时，通常都会触发 Shuffle Join，两表的所有分区节点会进行 All-to-All 的通讯，这种查询通常比较昂贵，会对网络 IO 会造成比较大的负担。

而对于大表和小表的连接操作，Spark 会在一定程度上进行优化，如果小表的数据量小于 Worker Node 的内存空间，Spark 会考虑将小表的数据广播到每一个 Worker Node，在每个工作节点内部执行连接计算，这可以降低网络的 IO，但会加大每个 Worker Node 的 CPU 负担。

是否采用广播方式进行 Join 取决于程序内部对小表的判断，如果想明确使用广播方式进行 Join，则可以在 DataFrame API 中使用 broadcast 方法指定需要广播的小表：

1 2	empDF.join(broadcast(deptDF), joinExpression).show() 复制代码

作者：heibaiying
链接：https://juejin.im/post/6844903950349500430
来源：掘金
著作权归作者所有。商业转载请联系作者获得授权，非商业转载请注明出处。

1.2 map-join小表误判

1 2	// a是一个几亿行的大表，b是一个只有几十行的小表。a和b都是由hive创建的表 select * from a where id not in (select id from b)

在spark ui中看到了该sql的执行计划，该sql语句执行了Map-side Join操作，但是spark把a表当成了小表，准备把a表broadcast到其他的节点，然后就是一直卡在这步broadcast操作上。造成上述问题的原因就是spark认为a表是一个小表，但是在spark ui上明显可以看到a表读了很多的行。但是为什么spark还会认为a表是一个小表呢？原因是spark判断一个hive表的大小会用hive的metastore数据来判断，因为我们的a表没有执行过ANALYZE TABLE，自然a表的metastore里面的数据就不准确了。

1.2.1 解决方法

设置spark.sql.statistics.fallBackToHdfs=True
该参数能让spark直接读取hdfs的文件大小来判断一个表达大小，从而代替从metastore里面的获取的关于表的信息。这样spark自然能正确的判断出表的大小，从而使用b表来进行broadcast。
使用hint
在使用sql语句执行的时候在sql语句里面加上mapjoin的注释，也能够达到相应的效果，比如把上述的sql语句改成:

1	select /+ BROADCAST (b) / * from a where id not in (select id from b)

这样spark也会使用b表来进行broadcast。

使用spark代码的方式
使用broadcast函数就能达到此效果：

1 2	from pyspark.sql.functions import broadcast broadcast(spark.table("b")).join(spark.table("a"), "id").show()

备注

只有当要进行join的表的大小小于spark.sql.autoBroadcastJoinThreshold（默认是10M）的时候，才会进行mapjoin。
Impala通过hint和执行表的位置调整也能够优化join操作，通过explain也可以查看sql的执行计划，然后再进行优化。

Join背景

当前SparkSQL支持三种join算法：Shuffle Hash Join、Broadcast Hash Join以及Sort Merge Join。其中前两者归根到底都属于Hash Join，只不过载Hash Join之前需要先Shuffle还是先Broadcast。其实，Hash Join算法来自于传统数据库，而Shuffle和Broadcast是大数据在分布式情况下的概念，两者结合的产物。因此可以说，大数据的根就是传统数据库。Hash Join是内核。

1.2.1.1 Spark Join的分类和实现机制

上图是Spark Join的分类和使用。

1.2.1.1.1 Hash Join

先来看看这样一条SQL语句：select * from order,item where item.id = order.i_id，参与join的两张表是order和item，join key分别是item.id以及order.i_id。现在假设Join采用的是hash join算法，整个过程会经历三步：

确定Build Table以及Probe Table：这个概念比较重要，Build Table会被构建成以join key为key的hash table，而Probe Table使用join key在这张hash table表中寻找符合条件的行，然后进行join链接。Build表和Probe表是Spark决定的。通常情况下，小表会被作为Build Table，较大的表会被作为Probe Table。
构建Hash Table：依次读取Build Table(item)的数据，对于每一条数据根据Join Key(item.id)进行hash，hash到对应的bucket中(类似于HashMap的原理)，最后会生成一张HashTable，HashTable会缓存在内存中，如果内存放不下会dump到磁盘中。
匹配：生成Hash Table后，在依次扫描Probe Table(order)的数据，使用相同的hash函数(在spark中，实际上就是要使用相同的partitioner)在Hash Table中寻找hash(join key)相同的值，如果匹配成功就将两者join在一起。

1.2.1.1.2 Broadcast Hash Join

当Join的一张表很小的时候，使用broadcast hash join。

Broadcast Hash Join的条件有以下几个：

被广播的表需要小于spark.sql.autoBroadcastJoinThreshold所配置的信息，默认是10M；
基表不能被广播，比如left outer join时，只能广播右表。

broadcast hash join可以分为两步：

broadcast阶段：将小表广播到所有的executor上，广播的算法有很多，最简单的是先发给driver，driver再统一分发给所有的executor，要不就是基于bittorrete的p2p思路；
hash join阶段：在每个executor上执行 hash join，小表构建为hash table，大表的分区数据匹配hash table中的数据。

1.2.1.1.3 Sort Merge Join

当两个表都非常大时，SparkSQL采用了一种全新的方案来对表进行Join，即Sort Merge Join。这种方式不用将一侧数据全部加载后再进行hash join，但需要在join前将数据进行排序。

首先将两张表按照join key进行重新shuffle，保证join key值相同的记录会被分在相应的分区，分区后对每个分区内的数据进行排序，排序后再对相应的分区内的记录进行连接。可以看出，无论分区有多大，Sort Merge Join都不用把一侧的数据全部加载到内存中，而是即用即丢；因为两个序列都有有序的，从头遍历，碰到key相同的就输出，如果不同，左边小就继续取左边，反之取右边。从而大大提高了大数据量下sql join的稳定性。

整个过程分为三个步骤：

shuffle阶段：将两张大表根据join key进行重新分区，两张表数据会分布到整个集群，以便分布式并行处理
sort阶段：对单个分区节点的两表数据，分别进行排序
merge阶段：对排好序的两张分区表数据执行join操作。join操作很简单，分别遍历两个有序序列，碰到相同join key就merge输出，否则继续取更小一边的key。

经过上文的分析，很明显可以得出这几种join的代价关系：cost(Broadcast Hash Join)< cost(Shuffle Hash Join) < cost(Sort Merge Join)，数据仓库设计时最好避免大表与大表的join查询，SparkSQL也可以根据内存资源、带宽资源适量将参数spark.sql. autoBroadcastJoinThreshold调大，让更多join实际执行为Broadcast Hash Join。

Guadazi

Spark join