Spark SQL中Lag函数返回NULL问题求助（Apache Iceberg场景）

你的猜想分析 #

你的猜想（分片过小导致lag()无法生成有效结果）不成立。Spark窗口函数的执行逻辑是：先根据PARTITION BY列对数据做shuffle，将同一分区的数据聚合到同一计算节点，再按ORDER BY排序后计算窗口函数。无论原始数据的分片（文件）数量多少，Spark都会保证每个分区内的数据完整，不会因分片大小导致窗口函数无法获取前置行。

可能的原因及排查步骤 #

1. 分区内数据量不足 #

lag(total, 12)表示取当前行往前第12行的total值，如果某个region+country分区下的date记录数不足12条，该分区内前11行的total_amount必然返回NULL。

排查SQL：

SELECT region, country, COUNT(DISTINCT date) AS date_count
FROM my_iceberg.temp_table_1
GROUP BY region, country
HAVING date_count < 12;

若返回结果，说明这些分区的数据量不足以支持lag(12)计算。

2. 日期列排序异常 #

如果date列是字符串类型（而非日期类型），排序会按字典序而非时间序执行，导致逻辑时间顺序混乱，lag()无法定位到正确的前置数据。

排查与修正：
先检查列类型：

DESCRIBE my_iceberg.temp_table_1;

若为字符串类型，需转换为日期类型后排序：

lag(total, 12) OVER (
    PARTITION BY region, country
    ORDER BY to_date(date)
) total_amount

3. Iceberg表元数据未刷新 #

执行rewrite_data_files后，Spark可能缓存旧的表元数据，导致读取的不是合并后的完整数据。

解决方式：
读取表前执行元数据刷新：

REFRESH TABLE my_iceberg.temp_table_1;

4. 分区列存在NULL值 #

region或country列的NULL值会被分到同一分区，若该分区的date记录数不足12条，对应total_amount会返回NULL。

排查SQL：

SELECT COUNT(*) AS null_partition_count
FROM my_iceberg.temp_table_1
WHERE region IS NULL OR country IS NULL;

5. 直接测试窗口函数结果 #

跳过插入temp_table_2的步骤，直接查询窗口函数结果，排除插入过程的问题：

SELECT
    date,
    region,
    country,
    total,
    lag(total, 12) OVER (
        PARTITION BY region, country
        ORDER BY date
    ) total_amount
FROM my_iceberg.temp_table_1
ORDER BY region, country, date;

观察NULL值对应的分区数据情况。

修正后的操作流程示例 #

-- 第一步：数据转换（补全GROUP BY，否则聚合函数语法错误）
INSERT INTO
    my_iceberg.temp_table_1 (
        date,
        region,
        country,
        total
    )
SELECT
    date,
    region,
    country,
    SUM(profit) as total
FROM
   my_iceberg.original_table
GROUP BY date, region, country;

-- 合并文件
CALL {spark_catalog}.system.rewrite_data_files(
    table => 'my_iceberg.temp_table_1',
    options => map(
        'min-input-files','1',
        'partial-progress.enabled','true'
    )
);

-- 刷新表元数据
REFRESH TABLE my_iceberg.temp_table_1;

-- 第二步：计算窗口函数并插入（补全目标表列，否则列数不匹配）
INSERT INTO
    my_iceberg.temp_table_2 (
        date,
        region,
        country,
        total,
        total_amount
    )
SELECT
    date,
    region,
    country,
    total,
    lag(total, 12) OVER (
        PARTITION BY region, country
        ORDER BY to_date(date)
    ) total_amount
FROM my_iceberg.temp_table_1;

注意：原查询存在两个语法错误：

1. 第一个INSERT的SELECT缺少GROUP BY date, region, country，聚合函数必须搭配分组；
2. 第二个INSERT的目标表temp_table_2定义了total_amount列，但原SELECT未包含该列，会导致列数不匹配。

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