具有当前行条件的 Spark 窗口函数

我正在尝试计算给定的order_id过去 365 天内有多少订单已付款。这不是问题：我使用窗函数 https://databricks.com/blog/2015/07/15/introducing-window-functions-in-spark-sql.html.

对我来说棘手的是：我不想在这个时间窗口内计算订单，因为payment_date是在之后order_date当前的order_id.

目前，我有这样的事情：

val window: WindowSpec = Window
  .partitionBy("customer_id")
  .orderBy("order_date")
  .rangeBetween(-365*days, -1)

and

df.withColumn("paid_order_count", count("*") over window)

这将统计客户当前订单之前过去 365 天内的所有订单。

现在我怎样才能为需要的计数添加一个条件order_date考虑当前订单？

Example:

+---------+-----------+-------------+------------+
|order_id |order_date |payment_date |customer_id |
+---------+-----------+-------------+------------+
|1        |2017-01-01 |2017-01-10   |A           |
|2        |2017-02-01 |2017-02-10   |A           |
|3        |2017-02-02 |2017-02-20   |A           |

结果表应如下所示：

+---------+-----------+-------------+------------+-----------------+
|order_id |order_date |payment_date |customer_id |paid_order_count |
+---------+-----------+-------------+------------+-----------------+
|1        |2017-01-01 |2017-01-10   |A           |0                |
|2        |2017-02-01 |2017-02-10   |A           |1                |
|3        |2017-02-02 |2017-02-20   |A           |1                |

For order_id = 3 the paid_order_count不应该是2 but 1 as order_id = 2是在之后支付的order_id = 3被放置。

我希望我能很好地解释我的问题并期待您的想法！

很好的问题！！！ 一些评论，使用范围之间创建一个基于其中的行数而不是值的固定框架，因此在两种情况下会出现问题：

1. 客户并非每天都有订单，因此 365 行窗口可能包含 order_date 早于一年前的行
2. 如果客户每天有多个订单，则会扰乱一年的保修范围
3. 1 和 2 的组合

Also 范围之间不适用于日期和时间戳数据类型。

为了解决这个问题，可以使用带有列表和 UDF 的窗口函数：

import org.apache.spark.sql.functions._
import org.apache.spark.sql.expressions.Window

  val df = spark.sparkContext.parallelize(Seq(
    (1, "2017-01-01", "2017-01-10", "A")
    , (2, "2017-02-01", "2017-02-10", "A")
    , (3, "2017-02-02", "2017-02-20", "A")
  )
  ).toDF("order_id", "order_date", "payment_date", "customer_id")
    .withColumn("order_date_ts", to_timestamp($"order_date", "yyyy-MM-dd").cast("long"))
    .withColumn("payment_date_ts", to_timestamp($"payment_date", "yyyy-MM-dd").cast("long"))

//      df.printSchema()
//      df.show(false)

  val window = Window.partitionBy("customer_id").orderBy("order_date_ts").rangeBetween(Window.unboundedPreceding, -1)

  val count_filtered_dates = udf( (days: Int, top: Long, array: Seq[Long]) => {
      val bottom = top - (days * 60 * 60 * 24).toLong // in spark timestamps are in secconds, calculating the date days ago
      array.count(v => v >= bottom && v < top)
    }
  )

  val res = df.withColumn("paid_orders", collect_list("payment_date_ts") over window)
      .withColumn("paid_order_count", count_filtered_dates(lit(365), $"order_date_ts", $"paid_orders"))

  res.show(false)

Output:

+--------+----------+------------+-----------+-------------+---------------+------------------------+----------------+
|order_id|order_date|payment_date|customer_id|order_date_ts|payment_date_ts|paid_orders             |paid_order_count|
+--------+----------+------------+-----------+-------------+---------------+------------------------+----------------+
|1       |2017-01-01|2017-01-10  |A          |1483228800   |1484006400     |[]                      |0               |
|2       |2017-02-01|2017-02-10  |A          |1485907200   |1486684800     |[1484006400]            |1               |
|3       |2017-02-02|2017-02-20  |A          |1485993600   |1487548800     |[1484006400, 1486684800]|1               |
+--------+----------+------------+-----------+-------------+---------------+------------------------+----------------+

将日期转换为 Spark 时间戳（以秒为单位）可以提高列表的内存效率。

这是最容易实现的代码，但不是最佳的，因为列表会占用一些内存，自定义 UDAF 是最好的，但需要更多编码，可能稍后会做。如果每个客户有数千个订单，这仍然有效。