无法理解 Spark 中的 UDF，尤其是 Java

Question

我正在尝试根据另一列的值在 Spark 数据集中创建一个新列。另一列的值在 json 文件中作为键进行搜索，并返回其值，即用于新列的值。

这是我尝试过的代码，但它不起作用，我不确定 UDF 的工作原理。在这种情况下如何使用 withColumn 或 udf 添加列？

Dataset<Row> df = spark.read().format("csv").option("header", "true").load("file path");
        Object obj = new JSONParser().parse(new FileReader("json path"));
        JSONObject jo = (JSONObject) obj;

        df = df.withColumn("cluster", functions.lit(jo.get(df.col("existing col_name")))));

任何帮助将不胜感激。提前致谢！

Answer 1

Spark 允许您使用 udf 函数创建自定义用户定义函数 (UDF)。

以下是如何定义 UDF 的 scala sn-p。

val obj = new JSONParser().parse(new FileReader("json path"));
val jo = obj.asInstanceOf[JSONObject];

def getJSONObject(key: String) = {
   jo.get(key)
}

定义函数后，可以将其转换为 UDF，如下所示：

 val getObject = udf(getJSONObject _)

有两种使用 UDF 的方法。

1. df.withColumn("cluster", lit(getObject(col("existing_col_name"))))

2. 如果你使用的是spark sql，你必须在使用前在sqlContext中注册你的udf。

   spark.sqlContext.udf.register("get_object", getJSONObject _)

   然后你就可以把它当作

   spark.sql("select get_object(existing_column) from some_table")

在这些中，使用哪个完全是主观的。

Answer 2

感谢@Constantine。从您的示例中，我能够更好地理解 UDF。这是我的java代码：

        Object obj = new JSONParser().parse(new FileReader("json path"));
        JSONObject jo = (JSONObject) obj;

        spark.udf().register("getJsonVal", new UDF1<String, String>() {
            @Override
            public String call(String key) {
                return  (String) jo.get(key.substring(0, 5));
            }
        }, DataTypes.StringType);

        df = df.withColumn("cluster", functions.callUDF("getJsonVal", df.col("existing col_name")));
        df.show(); // SHOWS NEW CLUSTER COLUMN