Spark中能否在Driver端设置Java UDF字段并在Executor端调用时使用？

可以实现，但你当前的写法存在问题，导致Driver端设置的字段值无法传递到Executor端。

问题原因 #

当你调用spark.udf.registerJavaFunction并传入UDF类名时，Spark会在每个Executor节点上重新实例化该UDF类，Driver端你初始化的jvm_udf对象仅存在于Driver本地，不会被序列化传递到Executor。因此Executor端的UDF实例始终使用默认的val = "foo"值。

解决方案 #

要让Driver端设置的字段值生效，需要将初始化好的UDF实例直接注册给Spark，而不是传递类名。具体步骤如下：

1. （可选）优化Java UDF代码（推荐）

如果允许修改Java UDF，添加带参数的构造函数来初始化字段，避免依赖init方法：

public class SomeUDF implements UDF2<String, String, String> {
    private String val;

    // 带参数构造函数，直接初始化val
    public SomeUDF(String st) {
        this.val = st;
    }

    // 保留无参构造（部分Spark版本要求必须存在）
    public SomeUDF() {
        this.val = "foo";
    }

    @Override
    public String call(String a, String b) {
        return val;
    }
}

2. 修改PySpark注册逻辑

直接使用Driver端初始化好的UDF实例进行注册，而不是传递类名：

from pyspark.sql.types import StringType
from pyspark.sql.functions import expr

# 创建已初始化的UDF实例
jvm_udf = spark._jvm.com.example.demo.SomeUDF("bla")
# 注册实例而非类名
spark.udf.register("foo", jvm_udf, StringType())

# 执行计算
df_single_row.withColumn("val", expr("foo('a','b')")).show()

这样Spark会将Driver端已初始化的UDF实例序列化后分发到所有Executor，Executor调用call方法时就会使用设置好的val = "bla"值，输出符合预期。

补充说明 #

如果无法修改Java UDF代码，也可以通过广播变量传递参数，但需要在Java UDF中获取广播变量的值（需确保UDF能访问SparkContext）。不过这种方式复杂度较高，优先推荐上述直接注册实例的方案。

内容的提问来源于stack exchange，提问作者YerivanLazerev