数据集上的慢联合

Question

我使用 Spark 2.1.1。我按小时循环对输入 DS (inputDs) 进行许多连接和选择，如下所示：

val myDs =  Iterator.iterate(fromDate)(_.plus(ofHours(1))).takeWhile(_.isBefore(toDate)).map(next => {
getDsForOneHour(inputDs, next.getYear, next.getMonthValue, next.getDayOfMonth, next.getHour)
}).reduce(_.union(_))

def getDsForOneHour(ds: Dataset[I], year:Int, month:Int, day:Int, hour: Int)(implicit sql: SQLImplicits):Dataset[I]= {
ds.where(col("year") === year and col("month") ===  month and col("day") ===  day and col("hour") === hour)
}

我使用 spark-testing-base 运行该代码，完成一个月的操作大约需要 3 分钟（约 30*24 个联合和选择）。这些都是懒惰的操作，我想知道为什么 Spark 构建 myDs 需要这么多时间？

Answer 1

我猜它很慢，因为执行计划会针对循环中联合的每个新数据集进行更新。您可以先重写代码来构建过滤器：

def getFilterForOneHour(year:Int, month:Int, day:Int, hour: Int): Column = {
  col("year") === year and col("month") ===  month and col("day") ===  day and col("hour") === hour
} 


val myFilter =  Iterator.iterate(fromDate)(_.plus(ofHours(1))).takeWhile(_.isBefore(toDate)).map(next => {
getFilterForOneHour(next.getYear, next.getMonthValue, next.getDayOfMonth, next.getHour)
}).reduce(_ or _)

val myDs = inputDs.where(myFilter)

编辑： 您还可以尝试进行分组联合（我的情况是批量大小为 50）。我已经运行了一些虚拟内存数据集的测试，在我的情况下，这将性能提高了 8 倍：

val myDs =  Iterator.iterate(fromDate)(_.plus(ofHours(1))).takeWhile(_.isBefore(toDate)).map(next => {
getDsForOneHour(inputDs, next.getYear, next.getMonthValue, next.getDayOfMonth, next.getHour)
})
.grouped(50).map(dss => dss.reduce(_ union _))
.reduce(_ union _)