Spark 迭代算法

Question

我们有一个用例，在 Spark 作业中

• 我们迭代外部表的分区
  • 加载该分区的数据（每个分区几乎相同的数据量）
  • 对数据进行转换（自连接，无udfs）
  • 将数据保存到外部位置
  • 重复

在此过程中，我们不缓存任何数据帧。

我们看到每次迭代后内存使用量稳步增加。一段时间后，作业退出并出现堆内存不足错误。

我正在寻找一些可能有助于调试此问题的指示。

代码看起来与此非常相似。

while(date < endDate) {
    val df = spark.sql("SELECT * FROM tbl JOIN tbl2 WHERE date = '${date}'")
    df.write.partitionBy("date").mode("overwrite").("s3://bucket/path")
    date = increment(date)
}