Spark SQL 可以引用上一个窗口/组的第一行吗？答案

【问题标题】：Can Spark SQL refer to the first row of the previous window / group?Spark SQL 可以引用上一个窗口/组的第一行吗？
【发布时间】：2021-04-21 03:39:17
【问题描述】：

我有一种看起来像这样的事件流：

Time UserId SessionId EventType EventData
1    2      A         Load      /a ...
2    1      B         Impressn  X ...
3    2      A         Impressn  Y ...
4    1      B         Load      /b ...
5    2      A         Load      /info ...
6    1      B         Load      /about ...
7    2      A         Impressn  Z ...

实际上，用户可以在较大的时间窗口内进行许多会话，并且还有一个点击事件类型，但在这里保持简单，我试图查看导致下一次加载的（页面浏览量）加载以及发生的印象聚合。

因此，在没有 SQL 的情况下，我已经加载了这个，按用户分组，按时间排序，并为每个会话用以前的加载信息（如果有的话）标记每一行。有一个

val outDS = logDataset.groupByKey(_.UserId)
      .flatMapGroups((_, iter) => gather(iter))

其中gather按时间对iter进行排序（可能是多余的，因为输入按时间排序），然后遍历序列，在每个新会话中将lastLoadData设置为null，将lastLoadData添加到每一行并将lastLoadData更新为如果该行是负载类型，则该行。产生类似的东西：

Time UserId SessionId EventType EventData  LastLoadData
1    2      A         Load      / ...      null
2    1      B         Impressn  X ...      null
3    2      A         Impressn  Y ...      / ...
4    1      B         Load      / ...      null
5    2      A         Load      /info ...  / ...
6    1      B         Load      /about ... / ...
7    2      A         Impressn  Z ...      /info ...

然后允许我汇总哪些（页面浏览量）加载会导致其他哪些加载，或者在每个（页面）加载哪些是前 5 个 Impressn 事件。

    outDS.createOrReplaceTempView(tempTable)
    val journeyPageViews = sparkSession.sql(
      s"""SELECT lastLoadData, EventData,
         | count(distinct UserId) as users,
         | count(distinct SessionId) as sessions
         |FROM ${tempTable}
         |WHERE EventType='Load'
         |GROUP BY lastLoadData, EventData""".stripMargin)

但是，我觉得添加 lastLoadData 列也可以使用 Spark SQL 窗口完成，但是我对其中的两个部分感到困惑：

如果我在 UserId+SessionId 上创建一个按时间排序的窗口，它如何应用于所有事件但查看前一个加载事件？（EG Impressn 将获得一个新列 lastLoadData 分配给此窗口的先前 EventData）
如果我以某种方式为每个会话的 Load 事件创建一个新窗口（也不确定如何），则窗口开头的 Load 事件（可能是“第一个”）应该获取上一个窗口的“第一个”的 lastLoadData，所以这可能是也不是正确的方法。

【问题讨论】：

标签： sql apache-spark apache-spark-sql window-functions

【解决方案1】：

您可以使用case when 使用null 屏蔽不是Load 的数据，并使用last 获取LastLoadData 并将ignorenull 设置为true：

logDataset.createOrReplaceTempView("table")
val logDataset2 = spark.sql("""
select 
    *,
    last(case when EventType = 'Load' then EventData end, true) 
        over (partition by UserId, SessionId
              order by Time
              rows between unbounded preceding and 1 preceding) LastLoadData 
from table
order by time
""")

logDataset2.show
+----+------+---------+---------+----------+------------+
|Time|UserId|SessionId|EventType| EventData|LastLoadData|
+----+------+---------+---------+----------+------------+
|   1|     2|        A|     Load|    /a ...|        null|
|   2|     1|        B| Impressn|     X ...|        null|
|   3|     2|        A| Impressn|     Y ...|      /a ...|
|   4|     1|        B|     Load|    /b ...|        null|
|   5|     2|        A|     Load| /info ...|      /a ...|
|   6|     1|        B|     Load|/about ...|      /b ...|
|   7|     2|        A| Impressn|     Z ...|   /info ...|
+----+------+---------+---------+----------+------------+

【讨论】：