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内存膨胀问题。
• 引用小表或子查询时，需要引用别名。

在下面这个表中，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;

可以通过如下三种方法解决。

• 手动切分热点值。

  将热点值分析出来后，从主表中过滤出热点值记录，先进行广播，再将剩余非热点值记录进行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