如何将 Spark 广播变量传递给 Java 中的 UDF？

Question

我有一个 Properties 变量，我通过调用 REST 服务来填充它。我需要将此列表传递给我的 UDF。我认为广播变量可以很好地满足我的目的，因为属性列表可能很长。所以我在我的主类中创建了一个广播变量：

Properties kp = getApplicationProperties(rootPath);
Broadcast<Properties> brVar = sc.broadcast(kp);

但是，我将在我的 UDF 中发送什么变量？我尝试使用 org.apache.spark.sql.functions.lit 发送文字值，但这导致我的 UDF 从未被调用：

spark.sqlContext().udf().registerJava("MongoInsert", MongoInsert.class.getName(), DataTypes.StringType);
persondatasetwithResid.select(callUDF("MongoInsert", lit(rootPath).cast(DataTypes.StringType),
                            col("value").cast(DataTypes.StringType), col("resourceId").cast(DataTypes.StringType),
                            lit(brVar))).show();


public class MongoInsert implements UDF4<String, String, String, Broadcast<Properties>, String> {
  public String call(String rootPath, String jsonstring, String resourceId, Broadcast<Properties> brVar)
        throws Exception {

    Properties kp = brVar.value();
  }
}

我找不到任何可以解释如何在 Java 中将广播变量传递给 UDF 的资源。请帮帮我。

编辑

我在网络上得到指示，只有列类型和文字字符串类型可以传递到 UDF。是这样吗？可以不传入其他变量，如地图、数组等吗？请记住，我说的是 Spark 和 Java。在 Scala 中，似乎有更多的灵活性。

编辑 2

我也收到了很多关于 typedLit 的文献，它实际上可以让我们使用 Maps 和 Arrays，例如。以下问题：

How to add a Map column to Spark dataset?

这是否意味着将变量包装在 lit() 中不会达到我的目的？我尝试用 typedLit() 包装一个简单的 Map 变量，但它给了我一个编译错误，说

"The method typedLit(T, TypeTags.TypeTag<T>) in the type functions is not applicable for the arguments (Map<String,String>)"

可以预见，网络上有大量关于如何在 Scala 中使用 typedLit 的资源，但在 Spark 和 Java 中几乎没有。

编辑 3

我发现了另一个问题：

how to set and get static variables from spark?

这个也可以为我提供答案，因为传递给所有类的静态变量可以满足我的目的。答案再次指定了广播变量，但也指定了闭包。再一次，在 Java 中没有使用这种闭包的例子，甚至在 Spark 官方文档中也没有！如果有人可以告诉我如何在 Java 中创建一个闭包并使用它将一个变量传递给 UDF，这将对我有很大帮助。

Answer 1

这是一个在类级别访问广播变量而不是被传入的示例。

import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.broadcast.Broadcast;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.api.java.UDF1;
import org.apache.spark.sql.types.DataTypes;

public class Test {

    Broadcast<String> broadcastVar = null;

    public UDF1 myudf =  new UDF1<String,String>(){
        @Override
        public String call(String x) {
            return broadcastVar.getValue();
        }
    };

    public void setBroadcastVar(Broadcast<String> broadcastVar){
        this.broadcastVar = broadcastVar;
    }


    public static void main(String[] args) {

        SparkSession spark = SparkSession.builder().master("local").appName("test").getOrCreate();
        JavaSparkContext js = new JavaSparkContext(spark.sparkContext());

        Test t = new Test();
        t.setBroadcastVar(js.broadcast("hellow world"));

        spark.udf().register("myudf",t.myudf,DataTypes.StringType);

       //use the udf

    }

}