在Spark Scala中定义UDF

Question

我需要使用Spark中的UDF，该UDF带有时间戳，整数和另一个dataframe，然后返回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 fordion int.type或对象dataFrame不是Package org.apache.parke.spark.sql

的成员

给出一些上下文：

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

我需要确定基于哪个是比视图日期更古老的价格输入的价格。

由于每个产品ID在第二个数据框中都有多个视图日期。我认为UDF比十字架更快。如果有人有不同的想法，我会很感激。

Answer 1

您无法在UDF 内传递 dataFrame，因为UDF将在特定分区的工作人员上运行。而且，由于您无法在Worker上使用RDD（是否可以在Apache Spark中创建嵌套的RDD？），同样，您也不能在Worker上使用DataFrame。

您需要为此进行工作！