PySpark中基于Timestamp粒度添加ML模型输入索引列

需求说明 #

为包含item_name、item_value、timestamp字段的表新增index列，要求相同timestamp对应相同的整数值索引，索引按timestamp升序从0开始编号。

支持窗口函数的数据库（MySQL 8.0+、PostgreSQL、SQL Server、Spark SQL等） #

使用DENSE_RANK()窗口函数，为每个唯一的timestamp分配连续索引，再减1实现从0开始编号：

SELECT 
    item_name,
    item_value,
    timestamp,
    DENSE_RANK() OVER (ORDER BY timestamp) - 1 AS index
FROM your_table_name;

说明：

• DENSE_RANK()会为相同的timestamp分配相同的排名，且排名连续无间隔
• 减1操作将默认从1开始的排名转换为从0开始的索引

MySQL 5.x（不支持窗口函数） #

使用用户变量实现索引分配：

SELECT 
    item_name,
    item_value,
    timestamp,
    @idx := CASE 
        WHEN @prev_ts = timestamp THEN @idx 
        ELSE @idx + 1 
    END - 1 AS index,
    @prev_ts := timestamp
FROM your_table_name,
     (SELECT @idx := 0, @prev_ts := NULL) AS vars
ORDER BY timestamp;

说明：

• 通过变量@prev_ts记录上一行的timestamp，@idx记录当前索引值
• 当当前行timestamp与上一行相同时，保持索引不变；否则索引加1
• 最终减1让索引从0开始

执行结果 #

运行上述SQL后，将得到符合期望的表结构：

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