使 spark-sql UDF 在 Scala spark 数据帧 DSL API 中可用答案

【问题标题】：Make spark-sql UDF available in Scala spark data frame DSL API使 spark-sql UDF 在 Scala spark 数据帧 DSL API 中可用
【发布时间】：2017-09-16 14:16:51
【问题描述】：

如何在 spark scala 数据框（非文本）api 中访问 geomesas UDF？ IE。如何转换

如何使 sql UDF 在文本 spark-sql API 中可用，在 scala 数据框架 DSL 中可用？ IE。如何启用而不是这个表达式

spark.sql("select st_asText(st_bufferPoint(geom,10)) from chicago where case_number = 1")

类似于

df.select(st_asText(st_bufferPoint('geom, 10))).filter('case_number === 1)

如何注册 geomesas UDF，使其不仅适用于 sql 文本模式。 SQLTypes.init(spark.sqlContext) from https://github.com/locationtech/geomesa/blob/f13d251f4d8ad68f4339b871a3283e43c39ad428/geomesa-spark/geomesa-spark-sql/src/main/scala/org/apache/spark/sql/SQLTypes.scala#L59-L66 似乎只注册文本表达式。

我已经在导入

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

所以这些功能

https://github.com/locationtech/geomesa/blob/828822dabccb6062118e36c58df8c3a7fa79b75b/geomesa-spark/geomesa-spark-sql/src/main/scala/org/apache/spark/sql/SQLSpatialFunctions.scala#L31-L41

应该可用。

【问题讨论】：

标签： scala apache-spark apache-spark-sql spark-dataframe geomesa

【解决方案1】：

您可以在要导入的 org.apache.spark.sql.functions 中使用 udf 函数，例如

val  myUdf = udf((x: String) => doSomethingWithX(x))

然后您可以在 DSL 中使用 myUdf，如 df.select(myUdf($"field"))

【讨论】：

但是 SQLTypes.init(spark.sqlContext) 已经在注册函数了。那么这可能是使用您的解决方案的问题吗？
是的，但你失去了参考，即。如果线是 val ST_DistanceSpheroid: (Geometry, Geometry) => jl.Double = nullableUDF((s, e) => fastDistance(s.getCoordinate, e.getCoordinate))。您需要在 DSL 中使用 ST_DistanceSpheroid

【解决方案2】：

看看org.apache.spark.sql.functions中的callUDF函数

val spark = SparkSession.builder()
  .appName("callUDF")
  .master("local[*]")
  .getOrCreate()
import spark.implicits._

val df = spark.createDataset(List("abcde", "bcdef", "cdefg")).toDF("str")
df.createTempView("view")

spark.sql("select length(substring(str, 2, 3)) from view").show()
df.select(callUDF("length", callUDF("substring", $"str", lit(2), lit(3)))).show()

spark.stop()

使用 Spark 2.1 测试

【讨论】：