Spark ML Transformer - 使用 rangeBetween 在窗口上聚合答案

【问题标题】：Spark ML Transformer - aggregate over a window using rangeBetweenSpark ML Transformer - 使用 rangeBetween 在窗口上聚合
【发布时间】：2017-11-03 15:20:00
【问题描述】：

我想创建自定义 Spark ML 转换器，该转换器在滚动窗口内应用聚合函数，构造为 over window。我希望能够在 Spark ML Pipeline 中使用这个转换器。

我想实现一些可以通过withColumn 很容易完成的事情，正如这个答案中给出的那样

Spark Window Functions - rangeBetween dates

例如：

val w = Window.orderBy(col("unixTimeMS")).rangeBetween(0, 700)
val df_new = df.withColumn("cts", sum("someColumnName").over(w))

在哪里

df 是我的数据框
unixTimeMS 是以毫秒为单位的 unix 时间
someColumnName 是我想要执行聚合的一些列。在此示例中，我对窗口中的行进行求和。
窗口w 包括当前事务和当前事务700 毫秒内的所有事务。

是否可以将这样的窗口聚合放入 Spark ML Transformer 中？

我能够使用 Spark ML SQLTransformer 实现类似的目标，其中

val query = """SELECT *,
              sum(someColumnName) over (order by unixTimeMS) as cts
              FROM __THIS__"""

new SQLTransformer().setStatement(query)

但我不知道如何在 SQL 中使用rangeBetween 来选择时间段。不仅仅是行数。我需要针对当前行的 unixTimeMS 的特定时间段。

我知道一元变换不是这样做的方法，因为我需要进行聚合。我是否需要定义一个 UDAF（用户定义的聚合函数）并在 SQLTransformer 中使用它？我找不到任何包含窗口函数的 UDAF 示例。

【问题讨论】：

标签： scala apache-spark-sql spark-dataframe window-functions apache-spark-ml

【解决方案1】：

我正在回答我自己的问题以供将来参考。我最终使用了 SQLTransformer。就像示例中的窗口函数一样，我使用了 range between：

val query = SELECT *,
sum(dollars) over (
      partition by Numerocarte
      order by UnixTime
      range between 1000 preceding and 200 following) as cts
      FROM __THIS__"

其中 1000 和 200 之间的范围与按列排序的单位有关。

【讨论】：