求助：如何将指定Athena SQL代码转换为PySpark SQL？

架构老林

2026-6-24

Athena SQL转PySpark SQL方案

原Athena SQL的核心逻辑是：按dt日期降序聚合生成包含dt和period_inactivity的数组，过滤出period_inactivity转整数后大于3的元素，取第一个符合条件的dt，并通过TRY处理空数组或索引越界的情况。以下是等价的PySpark SQL实现：

实现方式一（数组索引方式）

COALESCE(
    ELEMENT_AT(
        FILTER(
            ARRAY_AGG(ARRAY(dt, period_inactivity) ORDER BY date(dt) DESC),
            x -> CAST(x[1] AS INTEGER) > 3
        ),
        1
    )[0],
    NULL
) AS last_comeback

关键转换点说明：

索引差异：Athena（Presto）数组是1-based索引，PySpark数组默认是0-based。原代码中x[2]对应PySpark的x[1]（取period_inactivity），原代码取过滤后数组的第一个元素（[1]）对应PySpark的ELEMENT_AT(..., 1)（ELEMENT_AT支持1-based索引），再取该元素的dt对应[0]。
异常处理：Athena的TRY()用于捕获执行异常，PySpark中通过COALESCE配合ELEMENT_AT实现——当过滤后的数组为空或索引越界时，ELEMENT_AT返回null，最终COALESCE返回null。

实现方式二（UNNEST展开方式，更直观）

如果担心数组索引混淆，可使用UNNEST展开聚合数组后过滤取值：

COALESCE(
    (
        SELECT element.dt
        FROM UNNEST(
            ARRAY_AGG(STRUCT(dt, period_inactivity) ORDER BY date(dt) DESC)
        ) AS t(element)
        WHERE CAST(element.period_inactivity AS INTEGER) > 3
        LIMIT 1
    ),
    NULL
) AS last_comeback