Spark Scala中分区排序范围窗口的去重计数优化问询

问题背景 #

给定以下小时级时间粒度的DataFrame，需计算每个Group与TimeString（或UnixTimestamp）组合对应的过去12小时窗口内的DeviceId去重数量，窗口范围为当前时间点的前12小时（不含当前时间点）。

原始DataFrame：

+----------------------------------------------------+
|DeviceId | TimeString       | UnixTimestamp | Group |
+----------------------------------------------------+
|8313726  | 2023-10-17 01:00 | 1697529600    | g1    |
|0065322  | 2023-10-17 01:00 | 1697529600    | g2    |
|2902499  | 2023-10-17 00:00 | 1697526000    | g2    |
|1506928  | 2023-10-17 00:00 | 1697526000    | g1    |
|7466245  | 2023-10-17 14:00 | 1697576400    | g3    |
+------------------+----------------+----------------+

示例输出要求：

+----------------------------------------------------+
| Group | TimeString       | UniqueDevicesPast12Hours|
+----------------------------------------------------+
| g1    | 2023-10-17 00:00 |  1                   |
| g2    | 2023-10-17 01:00 |  2                   |
| g2    | 2023-10-17 00:00 |  1                   |
| g1    | 2023-10-17 01:00 |  2                   |
| g3    | 2023-10-17 14:00 |  1                   |
+------------------+----------------+----------------+

注：第一行表示g1组在2023-10-17 00:00之前12小时（2023-10-16 12:00至2023-10-16 23:00）内的唯一设备数。

已尝试方案及问题 #

方案1：UDF+窗口聚合 #

通过窗口函数收集12小时内的DeviceId，再用UDF计算去重数量，代码如下：

val windowSpecLast12 = Window.partitionBy("Group").
orderBy(col("UnixTimestamp").cast(LongType)). // 按数值时间戳排序
rangeBetween(-12 * 3600, -1) // 12小时的秒数范围

// 定义UDF计算数组内DeviceId的去重数量
val distinctCountUDF = udf((ids: Seq[String]) => ids.distinct.size)

df = df.withColumn("DistinctCountLast12", distinctCountUDF(collect_list("DeviceId").over(windowSpecLast12)))

问题：该方案逻辑可行，但大数据量下UDF性能极差，无法满足性能要求。

方案2：dense_rank替代去重计数 #

尝试用dense_rank()替代UDF计算去重数量，代码如下：

df = df.withColumn("RankLast12", dense_rank().over(windowSpecLast12))

问题：触发Spark分析异常：

org.apache.spark.sql.AnalysisException: Window Frame specifiedwindowframe(RangeFrame, -43200, -1) must match the required frame specifiedwindowframe(RowFrame, unboundedpreceding$(), currentrow$());

原因是dense_rank()这类排序窗口函数默认要求窗口帧为ROW BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW，与我们使用的RANGE类型窗口帧不兼容。

优化方案：Spark SQL高效实现 #

方案A：精确去重计算（自连接+分组聚合） #

通过自连接关联同一分组内12小时窗口内的数据，再分组统计去重设备数，完全利用Spark内置优化，避免UDF性能瓶颈。

Spark SQL代码：

WITH base_data AS (
    SELECT 
        `Group`,
        TimeString,
        UnixTimestamp,
        DeviceId
    FROM your_table_name -- 替换为你的表名或视图名
),
window_data AS (
    SELECT 
        b1.`Group`,
        b1.TimeString,
        b2.DeviceId
    FROM base_data b1
    INNER JOIN base_data b2
        ON b1.`Group` = b2.`Group`
        -- 关联条件：b2的时间在b1当前时间的前12小时到前1秒之间
        AND b2.UnixTimestamp BETWEEN (b1.UnixTimestamp - 12 * 3600) AND (b1.UnixTimestamp - 1)
)
SELECT 
    `Group`,
    TimeString,
    COUNT(DISTINCT DeviceId) AS UniqueDevicesPast12Hours
FROM window_data
GROUP BY `Group`, TimeString
ORDER BY `Group`, TimeString;

方案B：近似去重计算（窗口内置函数） #

如果业务允许一定误差（误差率约5%），可以使用Spark内置的approx_count_distinct窗口函数，性能远高于精确计算，且无需UDF。

Spark SQL代码：

SELECT 
    `Group`,
    TimeString,
    approx_count_distinct(DeviceId) OVER (
        PARTITION BY `Group`
        ORDER BY UnixTimestamp
        -- 窗口范围：当前时间前12小时到前1秒
        RANGE BETWEEN 12 * 3600 PRECEDING AND 1 PRECEDING
    ) AS UniqueDevicesPast12Hours
FROM your_table_name;

注：该方案依赖Spark版本对approx_count_distinct窗口函数的支持，建议使用Spark 3.0+版本。

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