在 Databricks 中从 UDF 内部查询 Delta Lake答案

【问题标题】：Querying Delta Lake from Inside of UDF in Databricks在 Databricks 中从 UDF 内部查询 Delta Lake
【发布时间】：2020-11-18 16:58:38
【问题描述】：

需要在结构化流中对 UDF 内的表执行一些查询。问题是，如果我尝试使用 spark.sql 在 UDF 内部，我会收到空指针异常。在这里遵循的最佳方式是什么。

基本上我需要从一个表中流式传输，然后使用该数据从另一个表执行一些范围查询。

例如。

val appleFilter = udf((appleId : String) => {
     val query = "select count(*) from appleMart where appleId='"+appleId+"'"
     val appleCount = spark.sql(query).collect().head.getLong(0)
     (appleCount>0)
})

val newApple = apples.filter(appleFilter($"appleId"))

【问题讨论】：

标签： apache-spark user-defined-functions databricks spark-structured-streaming delta-lake

【解决方案1】：

对于这项任务来说，这并不是一个真正正确的方法 - 您不应该从 UDF 内部进行单独的查询，因为 Spark 将无法并行化/优化它们。

更好的方法是在您的流数据帧和appleMart 数据帧之间创建一个join - 这将允许 Spark 优化所有操作。正如我从您的代码中了解到的那样，您只需要检查您是否有具有给定 ID 的苹果。在这种情况下，您可以只进行内部连接 - 这将只留下在 appleMart 中有行的 ID，如下所示：

val appleMart = spark.read.format("delta").load("path_to_delta")
val newApple = apples.join(appleMart, apples("appleId") === appleMart("appleId"))

如果由于某种原因您需要留下appleMart 中不存在的apples 条目，您可以改用left 连接...

附：如果appleMart 不经常更改，您可以缓存它。虽然，对于流式作业，对于查找表，例如 Cassandra could be better from performance standpoint。

【讨论】：