Spark SQL中是否有类似Vertica CONDITIONAL_CHANGE_EVENT的替代方案？

问题背景 #

你提到Vertica有一个CONDITIONAL_CHANGE_EVENT分析函数，用来标记字段值变化的事件，现在需要在Spark中实现类似逻辑：按Session_ID和Device_ID分区，按Channel_Time排序后，把连续相同的Channel归为同一个Group-ID，每次Channel变化时Group-ID递增。

原始数据 #

Session_ID,Device_ID,Channel_Time,Channel
1,1,4/9/2018 15:00:00,A
1,1,4/9/2018 15:01:00,A
1,1,4/9/2018 15:02:00,B
1,1,4/9/2018 15:03:00,B
1,1,4/9/2018 15:04:00,B
1,1,4/9/2018 15:05:00,C
1,1,4/9/2018 15:06:00,C
1,1,4/9/2018 15:07:00,A
1,1,4/9/2018 15:08:00,A
1,1,4/9/2018 15:09:00,B
1,1,4/9/2018 15:10:00,B

期望输出 #

Session_ID,Device_ID,Channel_Time,Channel,Group-ID
1,1,4/9/2018 15:00:00,A,1
1,1,4/9/2018 15:01:00,A,1
1,1,4/9/2018 15:02:00,B,2
1,1,4/9/2018 15:03:00,B,2
1,1,4/9/2018 15:04:00,B,2
1,1,4/9/2018 15:05:00,C,3
1,1,4/9/2018 15:06:00,C,3
1,1,4/9/2018 15:07:00,A,4
1,1,4/9/2018 15:08:00,A,4
1,1,4/9/2018 15:09:00,B,5
1,1,4/9/2018 15:10:00,B,5

你的当前实现 #

你用了4步SQL转换来实现，但过程比较繁琐，最终生成的Group-ID是基于行号的，不是连续的递增序列。

public class ConditionalTrueEvent { 
    public static void main(String[] args) { 
        SparkSession sparkSession = SparkSession.builder() 
            .appName(ConditionalTrueEvent.class.getName()) 
            .master("local[*]").getOrCreate(); 
        
        Dataset<Row> eventsDataSet = sparkSession.read() 
            .option("header", "true") 
            .csv("D:\\dev\\workspace\\java\\simple-kafka\\data\\test.csv"); 
        
        eventsDataSet.createOrReplaceTempView("rawView"); 
        sparkSession.sqlContext().sql("select * from rawView").show(); 
        
        Dataset<Row> channel_changed = sparkSession.sqlContext().sql("select * , " + 
            " row_number() over group_1 as row_number_by_session_device , " + 
            " (case when (lag(Channel,1,'XXX') over group_1 != Channel) then 1 else 0 end ) as channel_changed " + 
            " from rawView " + 
            "window group_1 as (partition by Session_ID , Device_ID order by Channel_Time )"); 
        channel_changed.createOrReplaceTempView("channel_changed"); 
        
        Dataset<Row> channel_changed_filled = sparkSession.sqlContext().sql("select * , " + 
            " ( case when channel_changed = 1 then row_number_by_session_device else 0 end ) as channel_changed_filled_row_num " + 
            " from channel_changed " + 
            "window group_1 as (partition by Session_ID , Device_ID order by Channel_Time )"); 
        channel_changed_filled.createOrReplaceTempView("channel_changed_filled"); 
        
        Dataset<Row> channel_changed_final = sparkSession.sqlContext().sql("select * , " + 
            " ( case when channel_changed_filled_row_num = 0 then max(channel_changed_filled_row_num) over group_1 else channel_changed_filled_row_num end ) as Group_ID " + 
            " from channel_changed_filled " + 
            "window group_1 as (partition by Session_ID , Device_ID order by Channel_Time )"); 
        channel_changed_final.createOrReplaceTempView("channel_changed_final"); 
        
        sparkSession.close(); 
    } 
}

更简便的实现方法 #

其实可以用窗口累加函数结合LAG函数一步到位，逻辑更清晰，代码更简洁：

核心SQL逻辑 #

SELECT 
    *,
    SUM(
        CASE 
            WHEN LAG(Channel, 1) OVER (PARTITION BY Session_ID, Device_ID ORDER BY Channel_Time) != Channel 
            THEN 1 
            ELSE 0 
        END
    ) OVER (PARTITION BY Session_ID, Device_ID ORDER BY Channel_Time) + 1 AS Group_ID
FROM rawView

逻辑解释 #

1. 标记变化事件：用LAG(Channel,1)获取当前行的上一行Channel值，比较是否和当前行不同，不同则标记为1，否则为0
2. 累加生成分组ID：对上述标记值在Session_ID+Device_ID分区内按时间顺序做累加求和，每次Channel变化时累加值会增加1，连续相同的Channel会共享同一个累加值
3. 调整起始值：最后加1是为了让Group-ID从1开始（第一行没有上一行，LAG返回null，标记为0，累加后是0，加1得到1）

简化后的Java代码 #

public class ConditionalTrueEvent { 
    public static void main(String[] args) { 
        SparkSession sparkSession = SparkSession.builder() 
            .appName(ConditionalTrueEvent.class.getName()) 
            .master("local[*]").getOrCreate(); 
        
        Dataset<Row> eventsDataSet = sparkSession.read() 
            .option("header", "true") 
            .csv("D:\\dev\\workspace\\java\\simple-kafka\\data\\test.csv"); 
        
        eventsDataSet.createOrReplaceTempView("rawView"); 
        
        // 简化的SQL逻辑，直接生成期望的Group-ID
        String simplifiedSql = """
            SELECT 
                *,
                SUM(
                    CASE 
                        WHEN LAG(Channel, 1) OVER (PARTITION BY Session_ID, Device_ID ORDER BY Channel_Time) != Channel 
                        THEN 1 
                        ELSE 0 
                    END
                ) OVER (PARTITION BY Session_ID, Device_ID ORDER BY Channel_Time) + 1 AS Group_ID
            FROM rawView
        """;
        
        Dataset<Row> result = sparkSession.sql(simplifiedSql);
        result.show();
        
        sparkSession.close(); 
    } 
}

执行这段代码后，就能得到你期望的连续递增的Group-ID，完全符合需求，而且比原来的4步转换简洁很多。

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