Apache Spark：如何在 Java 中通过数据集调用 UDF？答案

【问题标题】：Apache Spark: how to call UDF over dataset in Java?Apache Spark：如何在 Java 中通过数据集调用 UDF？
【发布时间】：2022-01-12 05:43:03
【问题描述】：

Java 中以下 Scala 代码 sn-p 的准确翻译是什么？

import org.apache.spark.sql.functions.udf 

def upper(s:String) : String = {
    s.toUpperCase
}
val toUpper = udf(upper _)
peopleDS.select(peopleDS(“name”), toUpper(peopledS(“name”))).show

请用Java填写以下缺失的语句：

import org.apache.spark.sql.api.java.UDF1;

UDF1 toUpper = new UDF1<String, String>() {
    public String call(final String str) throws Exception {
        return str.toUpperCase();
    }
};

peopleDS.select(peopleDS.col("name"), /* how to run  toUpper("name")) ? */.show();

注意

工作示例：

sqlContext.udf().register(
    "toUpper",
    (String s) -> s.toUpperCase(),
    DataTypes.StringType
);
peopleDF.selectExpr("toUpper(name)","name").show();

【问题讨论】：

标签： java scala apache-spark

【解决方案1】：

在 Java 中，没有注册就无法调用 UDF。请检查以下讨论：

Using UDFs in Java without registration

下面是你的 UDF：

private static UDF1 toUpper = new UDF1<String, String>() {
    public String call(final String str) throws Exception {
        return str.toUpperCase();
    }
};

注册UDF即可使用callUDF函数。

import static org.apache.spark.sql.functions.callUDF;
import static org.apache.spark.sql.functions.col;

sqlContext.udf().register("toUpper", toUpper, DataTypes.StringType);
peopleDF.select(col("name"),callUDF("toUpper", col("name"))).show();

【讨论】：

您的解决方案帮助我解决了我在使用 udf 时遇到的问题：org.apache.spark.SparkException: Task not serializable at org.apache.spark.util.ClosureCleaner$.ensureSerializable(ClosureCleaner.scala: 345）这发生在我的 udf 中，我使用了 Java 库中的 Cipher 类，我无法控制该类，也无法添加“实现可序列化”。

【解决方案2】：

Input csv:

+-------+--------+------+
|   name| address|salary|
+-------+--------+------+
|   Arun|  Indore|     1|
|Shubham|  Indore|     2|
| Mukesh|Hariyana|     3|
|   Arun|  Bhopal|     4|
|Shubham|Jabalpur|     5|
| Mukesh|  Rohtak|     6|
+-------+--------+------+

import org.apache.spark.SparkConf;
import org.apache.spark.SparkContext;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.functions;
import org.apache.spark.sql.api.java.UDF1;
import org.apache.spark.sql.types.DataTypes;

public static void main(String[] args) {
        SparkConf sparkConf = new SparkConf().setAppName("test").setMaster("local");
        SparkSession sparkSession = new SparkSession(new SparkContext(sparkConf));

        Dataset<Row> dataset = sparkSession.read().option("header", "true")
                .csv("C:\\Users\\Desktop\\Spark\\user.csv");

        /**Create udf*/
        UDF1<String, String> toLower = new UDF1<String, String>() {
            @Override
            public String call(String str) throws Exception {
                return str.toLowerCase();
            }
        };

        /**Register udf*/
        sparkSession.udf().register("toLower", toLower, DataTypes.StringType);

        /**call udf using functions.callUDF method*/
        dataset.select(dataset.col("name"),dataset.col("salary"), 
        functions.callUDF("toLower",dataset.col("address")).alias("address")).show();

}

Output :
+-------+------+--------+
|   name|salary| address|
+-------+------+--------+
|   Arun|     1|  indore|
|Shubham|     2|  indore|
| Mukesh|     3|hariyana|
|   Arun|     4|  bhopal|
|Shubham|     5|jabalpur|
| Mukesh|     6|  rohtak|
+-------+------+--------+

【讨论】：