在 Spark Scala 中定义 UDF

Question

我需要在 Spark 中使用一个 UDF，它接收一个时间戳、一个整数和另一个数据帧，并返回一个包含 3 个值的元组。

我不断地遇到一个又一个错误，我不确定我是否正在尝试修复它。

函数如下：

def determine_price (view_date: org.apache.spark.sql.types.TimestampType , product_id: Int, price_df: org.apache.spark.sql.DataFrame) : (Double, java.sql.Timestamp, Double) = {
    var price_df_filtered = price_df.filter($"mkt_product_id" === product_id && $"created"<= view_date)
    var price_df_joined = price_df_filtered.groupBy("mkt_product_id").agg("view_price" -> "min", "created" -> "max").withColumn("last_view_price_change", lit(1))
    var price_df_final = price_df_joined.join(price_df_filtered, price_df_joined("max(created)") === price_df_filtered("created")).filter($"last_view_price_change" === 1)
    var result = (price_df_final.select("view_price").head().getDouble(0), price_df_final.select("created").head().getTimestamp(0), price_df_final.select("min(view_price)").head().getDouble(0))
    return result
}
val det_price_udf = udf(determine_price)

它给我的错误是：

error: missing argument list for method determine_price
Unapplied methods are only converted to functions when a function type is expected.
You can make this conversion explicit by writing `determine_price _` or `determine_price(_,_,_)` instead of `determine_price`.

如果我开始添加参数，我会在其他错误中继续运行，例如 Int expected Int.type found 或 object DataFrame is not a member of package org.apache.spark.sql

给出一些上下文：

我的想法是我有一个价格数据框、一个产品 ID 和一个创建日期，以及另一个包含产品 ID 和查看日期的数据框。

我需要根据上次创建的比查看日期更早的价格条目来确定价格。

由于每个产品 ID 在第二个数据框中都有多个查看日期。我认为 UDF 比交叉连接更快。如果有人有不同的想法，我将不胜感激。

Answer 1

您不能在 UDF 中传递 Dataframe，因为 UDF 将在特定分区上的 Worker 上运行。由于你不能在 Worker 上使用 RDD（Is it possible to create nested RDDs in Apache Spark?），同样你也不能在 Worker 上使用 DataFrame。！

你需要解决这个问题！