Spark窗口分区功能需要永远完成

Question

给定一个数据框，我试图计算在过去 30 天内我看到 emailId 的次数。我的函数中的主要逻辑如下：

val new_df = df
  .withColumn("transaction_timestamp", unix_timestamp($"timestamp").cast(LongType))

val winSpec = Window
  .partitionBy("email")
  .orderBy(col("transaction_timestamp"))
  .rangeBetween(-NumberOfSecondsIn30Days, Window.currentRow)

val resultDF = new_df
  .filter(col("condition"))
  .withColumn("count", count(col("email")).over(winSpec))

配置：

spark.executor.cores=5

所以，我可以看到 5 个包含窗口函数的阶段，其中一些阶段完成得非常快（几秒钟内），还有 2 个阶段甚至在 3 小时内都没有完成，被卡在最后几个任务中（进展非常缓慢）：

这是数据倾斜的问题，如果我email从数据集中删除所有包含 5 个最高频率id的行，工作很快就会完成（不到 5 分钟）。

如果我尝试在 window partitionBy 中使用其他一些键，该作业将在几分钟内完成：

 Window.partitionBy("email", "date")

但显然，如果我这样做，它会执行错误的计数计算，这不是一个可接受的解决方案。

我尝试了其他各种火花设置，以增加内存、内核、并行度等，但这些设置似乎都没有帮助。

火花版本：2.2

当前的 Spark 配置：

-executor-memory：100G

-executor-cores: 5

-驱动内存：80G

-spark.executor.memory=100g

使用每台具有 16 核、128 GB 内存的机器。最大节点数高达 500。

解决这个问题的正确方法是什么？

更新：为了提供更多上下文，这里是原始数据框和相应的计算数据框：

 val df = Seq(
      ("a@gmail.com", "2019-10-01 00:04:00"),
      ("a@gmail.com", "2019-11-02 01:04:00"), 
      ("a@gmail.com", "2019-11-22 02:04:00"),
      ("a@gmail.com", "2019-11-22 05:04:00"),
      ("a@gmail.com", "2019-12-02 03:04:00"),
      ("a@gmail.com", "2020-01-01 04:04:00"),
      ("a@gmail.com", "2020-03-11 05:04:00"),
      ("a@gmail.com", "2020-04-05 12:04:00"),
      ("b@gmail.com", "2020-05-03 03:04:00")  
    ).toDF("email", "transaction_timestamp")


val expectedDF = Seq(
      ("a@gmail.com", "2019-10-01 00:04:00", 1),
      ("a@gmail.com", "2019-11-02 01:04:00", 1), // prev one falls outside of last 30 days win
      ("a@gmail.com", "2019-11-22 02:04:00", 2),
      ("a@gmail.com", "2019-11-22 05:04:00", 3),
      ("a@gmail.com", "2019-12-02 03:04:00", 3),
      ("a@gmail.com", "2020-01-01 04:04:00", 1),
      ("a@gmail.com", "2020-03-11 05:04:00", 1),
      ("a@gmail.com", "2020-04-05 12:04:00", 2),
      ("b@gmail.com", "2020-05-03 03:04:00", 1) // new email
).toDF("email", "transaction_timestamp", count") 

Answer 1

您的某些分区可能太大，这是因为某些电子邮件一个月内的数据太多。

要解决此问题，您可以创建一个仅包含电子邮件和时间戳的新数据框。然后，您按电子邮件和时间戳进行分组，计算行数并计算希望数据更少的窗口。如果时间戳容易重复，即如果df.count远大于，则计算速度将会加快df.select("email", "timestamp").distinct.count。如果不是这种情况，您可以截断时间戳，但会损失一些精度。这样，您就不必计算过去 30 天内的发生次数（误差一秒，因为时间戳以秒为单位），而是计算误差一分钟、一小时甚至一天的发生次数，具体取决于您的情况。需要。你会损失一点精度，但会大大加快计算速度。您提供的精度越高，获得的速度就越快。

代码如下所示：

// 3600 means hourly precision.
// Set to 60 for minute precision, 1 for second precision, 24*3600 for one day.
// Note that even precisionLoss = 1 might make you gain speed depending on
// the distribution of your data
val precisionLoss = 3600 
val win_size = NumberOfSecondsIn30Days / precisionLoss

val winSpec = Window
  .partitionBy("email")
  .orderBy("truncated_timestamp")
  .rangeBetween(-win_size, Window.currentRow)

val new_df = df.withColumn("truncated_timestamp",
                      unix_timestamp($"timestamp") / 3600 cast "long")

val counts = new_df
  .groupBy("email", "truncated_timestamp")
  .count
  .withColumn("count", sum('count) over winSpec)

val result = new_df
  .join(counts, Seq("email", "truncated_timestamp"))