Как передать широковещательную переменную Spark в UDF в Java?

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();
  }
}

Я не могу найти ресурс, который мог бы объясните, как передавать широковещательные переменные в UDF в Java. Пожалуйста, помогите мне.

РЕДАКТИРОВАТЬ

В Интернете я получаю указания, что в UDF могут передаваться только типы столбцов и литеральные типы строк. Это так? Могут ли другие переменные, такие как карты, массивы и т. Д. c. не будут переданы? Помните, я говорю о Спарк с Java. В Scala кажется, что это намного больше гибкости.

РЕДАКТИРОВАТЬ 2

Я также получаю много литературы, указывающей на то, что называется typedLit, который фактически позволяет нам работать с картами и массивами например следующий вопрос:

Как добавить столбец Map в набор данных Spark?

Означает ли это, что перенос переменной в lit () не будет служить моей цели? Я попытался обернуть простую переменную Map с помощью typedLit (), но она дает мне ошибку компиляции, говоря:

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

И вполне предсказуемо, в Интернете есть множество ресурсов о том, как использовать typedLit в Scala, но почти ничего в Spark с Java.

РЕДАКТИРОВАТЬ 3

Я нашел другой вопрос:

как установить и получить данные c переменные из spark?

Этот вопрос также может дать мне ответ, поскольку переменная stati c, передаваемая всем классам, может служить моей цели. Ответ снова указывает широковещательные переменные, но также определяет замыкания. Еще раз, нет примеров использования таких замыканий в 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

    }

}