Group By

一个带Group By的伪代码示例如下。

SELECT  shop_id
       ,sum(is_open) AS 营业天数
FROM    table_xxx_di
WHERE   dt BETWEEN '${bizdate_365}' AND '${bizdate}'
GROUP BY shop_id;

当发生数据倾斜时，可以通过如下两种方案解决：

方案一：添加随机数。

对SQL进行改写，添加随机数，把引起长尾的key进行拆分是解决Group By长尾的一种比较好的方法。

SELECT Key, Count(*) AS Cnt FROM TableName GROUP BY Key;

不考虑Combiner，M节点会Shuffle到R节点上，然后R节点再做Count操作，对应的执行计划是M->R。

假定已经找到了引起长尾的key，对长尾的key再做一次工作再分配，就变成：

-- 假设长尾的Key已经找到是KEY001
SELECT  a.Key
       ,SUM(a.Cnt) AS Cnt
FROM (
    SELECT  Key
           ,COUNT(*) AS Cnt
    FROM    TableName
    GROUP BY Key
           ,CASE WHEN Key = 'KEY001' THEN pmod(hash(rand()), 50)
                 ELSE 0
            END
) a
GROUP BY a.Key;

改完之后的执行计划变成了 M->R->R，虽然执行步骤变长了，但是长尾的key经过了 2 个步骤的处理，整体的时间消耗可能反而有所减少。与不加随机数的直接聚合相比，资源消耗与耗时效果基本持平；但实际场景中引发长尾的key不止一个，再考虑寻找长尾key和SQL改写的投入成本，直接对长尾key添加随机数（不做两阶段改写）的投入成本会更低一些。

方案二：创建滚存表。

降本提效，核心诉求是取过去一年的商户数据，对于线上任务而言，每次都要读取T-1至T-365的所有分区其实是对资源的很大浪费，创建滚存表可以减少分区的读取但是又不影响过去一年的取数，示例如下。

首次初始化365天的商户营业数据（Group By汇总），标记数据更新日期，记为表a；后续线上任务切换为T-2日表a关联table_xxx_di表再Group By，这样每天读取的数据从365减少到了2个，主键shopid的重复性极大降低，对资源的消耗也会减少。

-- 创建滚存表
CREATE TABLE IF NOT EXISTS m_xxx_365_df (
   shop_id STRING COMMENT '商户ID',
   last_update_ds STRING COMMENT '最近更新日期',
   `365d_open_days` BIGINT COMMENT '365天营业天数'
)
PARTITIONED BY (ds STRING COMMENT '日期分区');

-- 假定 365d 是 2021.5.1-2022.5.1，先完成一次初始化
INSERT OVERWRITE TABLE m_xxx_365_df PARTITION(ds = '20220501')
SELECT shop_id
      ,max(ds) AS last_update_ds
      ,sum(is_open) AS `365d_open_days`
FROM table_xxx_di
WHERE ds BETWEEN '20210501' AND '20220501'
GROUP BY shop_id;

-- 那么之后线上任务要执行的是
INSERT OVERWRITE TABLE m_xxx_365_df PARTITION(ds = '${bizdate}')
SELECT aa.shop_id
      ,aa.last_update_ds
      ,`365d_open_days` - COALESCE(is_open, 0) AS `365d_open_days`  -- 消除营业天数的无限滚存
FROM (
    SELECT shop_id
          ,max(last_update_ds) AS last_update_ds
          ,sum(`365d_open_days`) AS `365d_open_days`
    FROM (
        SELECT shop_id
              ,ds AS last_update_ds
              ,sum(is_open) AS `365d_open_days`
        FROM table_xxx_di
        WHERE ds = '${bizdate}'
        GROUP BY shop_id
        UNION ALL
        SELECT shop_id
              ,last_update_ds
              ,`365d_open_days`
        FROM m_xxx_365_df
        WHERE ds = '${bizdate_2}' AND last_update_ds >= '${bizdate_365}'
        GROUP BY shop_id
    )
    GROUP BY shop_id
) AS aa
LEFT JOIN (
    SELECT shop_id, is_open
    FROM table_xxx_di
    WHERE ds = '${bizdate_366}'
) AS bb
ON aa.shop_id = bb.shop_id;