Spark-sql Join优化=＞(cache+BroadCast)_spark cache join wrong

#Spark-sql Join优化
##背景

spark-sql或者hive-sql 很多业务场景都会有表关联的的操作，在hive中有map side join优化，对应的在spark-sql中也有map side join。spark中如果在参与join的表中存在小表，可以采用cache broadcast的方式进行优化，避免数据的shuffle，从而一定程度上可以避免数据倾斜，增加spark作业的执行速度。
　　本文主要阐述怎么使用spark sql的map side join进行优化，及使用过程需要注意的内容，同时mark自己研究spark的过程。
　　　　　　　　　　　　　　　　　　　　　　　　　　 　　　　　　　　　　　　　　　　　　　 ——　 [ 败八 ]

##避免使用shuffle类型的操作

如果有可能的话，要尽量避免使用shuffle类算子。因为Spark作业运行过程中，最消耗性能的地方就是shuffle过程。shuffle过程，简单来说，就是将分布在集群中多个节点上的同一个key，拉取到同一个节点上，进行聚合或join等操作。比如reduceByKey、join等算子，都会触发shuffle操作。

shuffle过程中，各个节点上的相同key都会先写入本地磁盘文件中，然后其他节点需要通过网络传输拉取各个节点上的磁盘文件中的相同key。而且相同key都拉取到同一个节点进行聚合操作时，还有可能会因为一个节点上处理的key过多，导致内存不够存放，进而溢写到磁盘文件中。因此在shuffle过程中，可能会发生大量的磁盘文件读写的IO操作，以及数据的网络传输操作。磁盘IO和网络数据传输也是shuffle性能较差的主要原因。

因此在我们的开发过程中，能避免则尽可能避免使用reduceByKey、join、distinct、repartition等会进行shuffle的算子，尽量使用map类的非shuffle算子。这样的话，没有shuffle操作或者仅有较少shuffle操作的Spark作业，可以大大减少性能开销。

举例分析

ipTable：需要进行关联的几千条ip数据(70k)和hist：历史数据（百亿级别）
　　直接join将会对所有数据进行shuffle，需要大量的io操作，相同的key会在同一个partition中进行处理，任务的并发度也收到了限制。

使用broadcast将会把小表分发到每台执行节点上，因此，关联操作都在本地完成，基本就取消了shuffle的过程，运行效率大幅度提高。

实践

###测试代码

 val spark = SparkSession
      .builder()
      .appName("broadcast join test")
      .config("hive.exec.scratchdir", s"/user/$user/hive-$user")
      .config("spark.sql.warehouse.dir", s"/user/$user/warehouse")
      .enableHiveSupport()
      .getOrCreate()
//使用broadcast join 首先要cache表
val table_a= sql(
      s"""
         |CACHE TABLE table_a AS
         | select
         |            value,
         |            max(name) as name
         |        from
         |            table_test
         |        where
         |            dt='${date}'
         |        group by
         |            value
      """.stripMargin)
      
  val sql_strs =
      |s"""
      |  select * from
      |    table_b as ti
      |  left outer join table_a as t3
      |  on t1.interest_id=t3.value

       sql(sql_strs)
       spark.stop()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

注意

• cache 的表不一定会被广播到Executor，执行map side join！！！
• 有另外一个参数：spark.sql.autoBroadcastJoinThreshold 会判断是否将该表广播；
• spark.sql.autoBroadcastJoinThreshold参数官方解释：

  Configures the maximum size in bytes for a table that will be broadcast to all worker nodes when performing a join. By setting this value to -1 broadcasting can be disabled. Note that currently statistics are only supported for Hive Metastore tables where the command

• spark.sql.autoBroadcastJoinThreshold参数默认值是10M，所以只有cache的表小于10M的才被广播到Executor上去执行map side join，因此要特别要注意，因此在选择cache表的时候，要注意表的大小和spark.sql.autoBroadcastJoinThreshold参数的调整。如果内存比较充足，建议调大该参数。

reference

• http://www.jianshu.com/p/2c7689294a73
• http://tech.meituan.com/spark-tuning-basic.html