# Join
针对Join端产生的数据倾斜，会存在多种不同的情况，例如大表和小表Join、大表和中表Join、Join热点值长尾。
#### 大表Join小表
- **数据倾斜示例**
  如下示例中t1是一张大表，t2、t3是小表。
  ```
  SELECT  t1.ip
         ,t1.is_anon
         ,t1.user_id
         ,t1.user_agent
         ,t1.referer
         ,t2.ssl_ciphers
         ,t3.shop_province_name
         ,t3.shop_city_name
  FROM    viewtable t1
  LEFT OUTER JOIN other_viewtable t2
         ON t1.header_eagleeye_traceid = t2.eagleeye_traceid
  LEFT OUTER JOIN (
             SELECT  shop_id
                    ,city_name AS shop_city_name
                    ,province_name AS shop_province_name
             FROM    tenanttable
             WHERE   ds = (SELECT max(ds) FROM tenanttable)
             AND     is_valid = 1
         ) t3
         ON t1.shopid = t3.shop_id
  ```
  
- **解决方案**
  使用Hint（查询优化提示指令）语法，广播小表，如下所示。
  ```
  SELECT  /*+ BROADCAST(t2, t3) */
          t1.ip
         ,t1.is_anon
         ,t1.user_id
         ,t1.user_agent
         ,t1.referer
         ,t2.ssl_ciphers
         ,t3.shop_province_name
         ,t3.shop_city_name
  FROM    viewtable t1
  LEFT OUTER JOIN other_viewtable t2
         ON t1.header_eagleeye_traceid = t2.eagleeye_traceid
  LEFT OUTER JOIN (
             SELECT  shop_id
                    ,city_name AS shop_city_name
                    ,province_name AS shop_province_name
             FROM    tenanttable
             WHERE   ds = (SELECT max(ds) FROM tenanttable)
             AND     is_valid = 1
         ) t3
         ON t1.shopid = t3.shop_id
  ```
  
- **注意事项**
  - Spark支持自动触发广播，阈值配置参数为spark.sql.autoBroadcastJoinThreshold=10485760（默认值为10MB）。手动Hint则不受阈值参数限制，需要注意广播引起的Driver/Executor内存膨胀问题。
  
  - 引用小表或子查询时，需要引用别名。
   
 
#### Join热点值长尾
- **数据倾斜示例**
  在下面这个表中，eleme_uid中存在很多热点数据，容易发生数据倾斜。
  ```
  SELECT
      eleme_uid,
      ...
  FROM (
      SELECT
          eleme_uid,
          ...
      FROM <viewtable>
  ) t1
  LEFT JOIN (
      SELECT
          eleme_uid,
          ...
      FROM <customertable>
  ) t2
  ON t1.eleme_uid = t2.eleme_uid;
  ```
  
- **解决方案**
  可以通过如下三种方法解决。
  
  | **序号** | **方案**                         | **说明**                                                                            |
  |:---|:---|:---|
  | 方案一     | 手动切分热点值                         | 将热点值分析出来后，从主表中过滤出热点值记录先进行广播（Broadcast Join），剩余非热点记录走Sort Merge Join，最后UNION ALL。 |
  | 方案二     | AQE Skew Join / SkewJoin Hint | AQE自动检测倾斜key拆分；也可手动用 /\*+ SKEW \*/ Hint。                                           |
  | 方案三   | 倍数表取模相等Join                    | 利用倍数表将热点值放大N倍，然后Join时使用用户ID和number两个关联键。                                           |
     
  - 手动切分热点值。 将热点值分析出来后，从主表中过滤出热点值记录，先进行广播，再将剩余非热点值记录进行Sort Merge Join，最后合并两部分的Join结果。具体可以参考如下代码示例：
    ```
    -- 热点值走Broadcast Join
    SELECT /*+ BROADCAST(t2) */
           eleme_uid, ...
    FROM (SELECT eleme_uid, ... FROM viewtable WHERE eleme_uid = <skewed_value>) t1
    LEFT JOIN (SELECT eleme_uid, ... FROM customertable WHERE eleme_uid = <skewed_value>) t2
           ON t1.eleme_uid = t2.eleme_uid
    UNION ALL
    -- 非热点值走Sort Merge Join
    SELECT eleme_uid, ...
    FROM (SELECT eleme_uid, ... FROM viewtable WHERE eleme_uid != <skewed_value>) t3
    LEFT JOIN (SELECT eleme_uid, ... FROM customertable WHERE eleme_uid != <skewed_value>) t4
           ON t3.eleme_uid = t4.eleme_uid
    ```
    
  
  - AQE Skew Join / SkewJoin Hint。 优先推荐开启AQE Skew Join，配置参数如下：
    ```
    spark.sql.adaptive.enabled=true
    spark.sql.adaptive.skewJoin.enabled=true
    ```
    Spark也支持手动Hint（查询优化提示指令）指定倾斜值，使用示例如下：
    ```
    SELECT /*+ SKEW('a', 'c0', (1, '2'), (3, '4')) */ *
    FROM T0 a JOIN T1 b
      ON a.c0 = b.c0 AND a.c1 = b.c1 AND a.c2 = b.c2;
    ```
    支持的Join类型：
    - Inner Join可以对Join两侧表中的任意一侧进行Hint。
    
    - Left Join、Semi Join和Anti Join只可以Hint左侧表。
    
    - Right Join只可以Hint右侧表。
    
    - Full Join不支持Skew Join Hint。
    
    
    建议只对一定会出现数据倾斜的Join添加Hint，因为Hint会运行一个Aggregate，存在一定代价。Join key类型必须一致（两侧字段类型相同），否则Hint不生效；可通过CAST在子查询中统一类型。
    
  
  - 倍数表取模相等Join 该方案和前两个方案的逻辑不同，不是分而治之的思路，而是利用一个倍数表，其值只有一列：int列，比如可以是从1到N（具体可根据倾斜程度确定），利用这个倍数表可以将用户行为表放大N倍，然后Join时使用用户ID和number两个关联键。这样原先只按照用户ID分发导致的数据倾斜就会由于加入了number关联条件而减少为原先的1/N。但是这样做也会导致数据膨胀N倍。
    ```
    SELECT eleme_uid, ...
    FROM (SELECT eleme_uid, ... FROM viewtable) t1
    LEFT JOIN (SELECT /*+ BROADCAST(multipletable) */
                      eleme_uid, number, ...
               FROM customertable JOIN multipletable) t2
           ON t1.eleme_uid = t2.eleme_uid
          AND mod(t1.value_col, 10) + 1 = t2.number
    ```