Impala中Iceberg表使用INSERT OVERWRITE+ROW_NUMBER()去重无效

问题原因 #

核心问题在于分区级别的INSERT OVERWRITE只会覆盖查询结果中涉及的分区，原表中未被查询结果覆盖的分区内的旧重复记录会被保留。

比如你的数据里，C1的两条记录分别在created_at=2024-01-01和created_at=2024-01-02两个分区。去重逻辑筛选出的是modified_at最新的2024-01-01分区记录，执行INSERT OVERWRITE ... PARTITION(created_at)时，只会覆盖这个分区，而2024-01-02分区里的旧C1记录并没有被处理，所以表中仍存在重复。

而写入新表时，新表初始为空，所有去重后的记录全量写入，自然不会有残留旧数据，因此结果正常。

解决方法 #

方法1：全量覆盖整个表（简单直接） #

去掉PARTITION(created_at)子句，让INSERT OVERWRITE直接替换整个Iceberg表的内容，所有旧数据都会被去重后的结果替代：

INSERT OVERWRITE TABLE customer_fact
WITH ordered AS (
    SELECT *,
           ROW_NUMBER() OVER (PARTITION BY customer_id ORDER BY modified_at DESC) AS rn
    FROM customer_fact
)
SELECT
 customer_id,
 created_at,
 modified_at,
 status_flag 
FROM ordered
WHERE rn = 1;

注意：这种方式会重写整个表，超大规模表可能有性能开销，但逻辑最简单可靠。

方法2：先清理冗余分区再插入（适合大表） #

如果表数据量极大，全量覆盖代价过高，可分两步操作：

1. 先找出并删除包含冗余记录的分区：

-- 找出所有包含冗余记录的分区
WITH customer_partitions AS (
    SELECT customer_id, COUNT(DISTINCT created_at) AS partition_count
    FROM customer_fact
    GROUP BY customer_id
    HAVING partition_count > 1
),
redundant_partitions AS (
    SELECT cf.created_at
    FROM customer_fact cf
    JOIN customer_partitions cp ON cf.customer_id = cp.customer_id
    WHERE cf.modified_at != (SELECT MAX(modified_at) FROM customer_fact WHERE customer_id = cf.customer_id)
)
ALTER TABLE customer_fact DROP IF EXISTS PARTITION (created_at) IN (SELECT created_at FROM redundant_partitions);

2. 再执行原去重插入逻辑，确保保留的分区内数据是最新的：

INSERT OVERWRITE TABLE customer_fact PARTITION(created_at)
WITH ordered AS (
    SELECT *,
           ROW_NUMBER() OVER (PARTITION BY customer_id ORDER BY modified_at DESC) AS rn
    FROM customer_fact
)
SELECT
 customer_id,
 created_at,
 modified_at,
 status_flag 
FROM ordered
WHERE rn = 1;

这种方式只清理有冗余的分区，减少不必要的数据重写。

验证 #

执行完操作后，查询表数据：

SELECT * FROM customer_fact;

应仅返回每个customer_id的最新记录：

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