Hive使用collect_set生成用户产品子集时缺失数据的问题排查

我在Hive中有如下orders表：

CREATE TABLE orders (
customer_id INT,
date DATE,
product_id INT
);

INSERT INTO orders VALUES
(1, '2024-02-18', 101),
(1, '2024-02-18', 102),
(1, '2024-02-19', 101),
(1, '2024-02-19', 103),
(2, '2024-02-18', 104),
(2, '2024-02-18', 105),
(2, '2024-02-19', 101),
(2, '2024-02-19', 106);

希望得到每个日期下每个用户的所有产品子集，预期结果如下：

date       | partition_set   |
------------|-----------------|
2024-02-18 | [101]           |
2024-02-18 | [101, 102]      |
2024-02-18 | [102]           |
2024-02-18 | [104]           |
2024-02-18 | [104, 105]      |
2024-02-18 | [105]           |
2024-02-19 | [101]           |
2024-02-19 | [101, 103]      |
2024-02-19 | [103]           |
2024-02-19 | [106]           |
2024-02-19 | [101, 106]      |

我尝试了以下查询：

select `date`, customer_id , COLLECT_SET(TRIM(upper(product_id)))  
over(PARTITION by  `date`, customer_id ORDER by customer_id,upper(product_id)) prd
from reporting.tmp_orders;

得到的结果中，用户1在2024-02-18的结果缺少了[102]这一行，请问我的查询哪里出错了？需要用Hive实现该需求，而非Python。

你当前的查询用了窗口函数COLLECT_SET()加ORDER BY，这种写法会生成累积的集合，而不是所有可能的子集。以用户1在2024-02-18的数据为例：

• 第一条记录（product_id=101）生成[101]
• 第二条记录（product_id=102）生成[101,102]
  但窗口函数只会为每一行生成对应位置的累积集合，不会单独生成只包含102的集合，所以自然缺少[102]这一行。

另外，TRIM(upper(product_id))是多余的——product_id是INT类型，转大写、修剪字符串操作毫无意义，反而可能引发类型转换问题。

要生成每个用户每日所有产品的非空子集，这里提供两种可行的实现方式：

方式一：自关联生成子集（适合产品数量较多的场景） #

WITH user_daily_products AS (
    -- 为每个用户每日的产品分配序号
    SELECT 
        `date`,
        customer_id,
        product_id,
        ROW_NUMBER() OVER(PARTITION BY `date`, customer_id ORDER BY product_id) AS rn
    FROM orders
),
product_combinations AS (
    -- 自关联生成连续的产品组合
    SELECT 
        a.`date`,
        COLLECT_LIST(b.product_id) AS partition_set
    FROM user_daily_products a
    JOIN user_daily_products b 
        ON a.`date` = b.`date` 
        AND a.customer_id = b.customer_id 
        AND b.rn >= a.rn
    GROUP BY a.`date`, a.customer_id, a.rn
)
-- 去重并排序输出
SELECT DISTINCT
    `date`,
    partition_set
FROM product_combinations
ORDER BY `date`, partition_set;

方式二：位运算生成子集（适合产品数量较少的场景） #

WITH user_daily_product_lists AS (
    -- 先聚合每个用户每日的产品列表
    SELECT 
        `date`,
        customer_id,
        COLLECT_LIST(product_id) AS product_list
    FROM orders
    GROUP BY `date`, customer_id
),
subset_generation AS (
    -- 利用位运算生成所有非空子集
    SELECT 
        `date`,
        COLLECT_LIST(product_list[pos]) AS partition_set
    FROM user_daily_product_lists
    LATERAL VIEW posexplode(product_list) pe AS pos, product
    LATERAL VIEW generate_series(1, pow(2, SIZE(product_list)) - 1) gs AS num
    WHERE BITGET(num, pos) = 1
    GROUP BY `date`, customer_id, num
)
-- 输出结果
SELECT 
    `date`,
    partition_set
FROM subset_generation
ORDER BY `date`, partition_set;

内容的提问来源于stack exchange，提问作者Param