Spark Connect使用UDF报错‘SparkContext或SparkSession需先创建’求助

问题详情 #

已启动Spark Connect服务器，不使用UDF时操作正常（如df.show()可正常执行），但使用UDF时抛出**"SparkContext or SparkSession should be created first"**错误。SparkSession已确认存在（可创建并打印DataFrame），仅在应用UDF时失败。疑问：Spark Connect是否暂不支持UDF或依赖SparkContext的功能？

复现步骤 #

• 启动Spark Connect服务器（可参考官方文档或Windows环境配置方法）
• 执行pytest -k test_spark_connect在远程会话测试UDF
• 执行pytest -k test_spark在本地Spark会话执行相同测试（注：同时运行两个测试会冲突，需分开执行）

测试代码 #

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, udf

@udf
def to_upper(s):
    if s is not None:
        return s.upper()

def test_spark_connect():
    simple_spark_test(SparkSession.builder.remote("sc://localhost:15002").getOrCreate())

def test_spark():
    simple_spark_test(SparkSession.builder.getOrCreate())

def simple_spark_test(session_to_test: SparkSession):
    print(f'\nss: {session_to_test}')
    print(f'ss.conf.get("spark.app.name"): {session_to_test.conf.get("spark.app.name")}')

    df = session_to_test.createDataFrame([(1, "John Doe")], ("id", "name"))
    df.show()
    df.select(col("name"), to_upper("name")).show()

测试结果 #

• 远程会话测试结果：执行失败，抛出RuntimeError: SparkContext or SparkSession should be created first
• 本地会话测试结果：执行成功，UDF正常转换字符串为大写

Spark Connect 目前不支持传统的Python UDF，核心原因是传统UDF依赖本地的SparkContext/SparkSession实例，而Spark Connect采用客户端-服务器分离架构，客户端的SparkSession仅为远程代理，不存在本地SparkContext实例，这就是触发错误的根源。

要在Spark Connect中实现自定义函数逻辑，需使用其兼容的API方案：

方案1：注册远程UDF #

将自定义函数注册到远程Spark会话，通过SQL表达式调用：

from pyspark.sql import SparkSession
from pyspark.sql.functions import col
from pyspark.sql.types import StringType

def to_upper(s):
    if s is not None:
        return s.upper()

def test_spark_connect():
    spark = SparkSession.builder.remote("sc://localhost:15002").getOrCreate()
    # 注册UDF到远程会话
    spark.udf.register("to_upper", to_upper, StringType())
    simple_spark_test(spark)

def test_spark():
    spark = SparkSession.builder.getOrCreate()
    from pyspark.sql.functions import udf
    to_upper_udf = udf(to_upper, StringType())
    simple_spark_test(spark, to_upper_udf)

def simple_spark_test(session_to_test: SparkSession, udf_func=None):
    print(f'\nss: {session_to_test}')
    print(f'ss.conf.get("spark.app.name"): {session_to_test.conf.get("spark.app.name")}')

    df = session_to_test.createDataFrame([(1, "John Doe")], ("id", "name"))
    df.show()
    if udf_func:
        # 本地会话使用传统UDF调用方式
        df.select(col("name"), udf_func(col("name"))).show()
    else:
        # Spark Connect使用注册的UDF，通过SQL表达式调用
        df.select(col("name"), session_to_test.sql_expression("to_upper(name)")).show()

方案2：使用Pandas UDF（Spark 3.4+） #

若服务器端已安装Pandas和PyArrow，可使用Pandas UDF，它在Spark Connect中是兼容的：

from pyspark.sql.functions import pandas_udf
from pyspark.sql.types import StringType
import pandas as pd

@pandas_udf(StringType())
def to_upper_pandas(s: pd.Series) -> pd.Series:
    return s.str.upper()

# 在Spark Connect会话中直接调用即可
df.select(col("name"), to_upper_pandas(col("name"))).show()

注意事项 #

• Spark Connect对UDF的支持从Spark 3.3版本开始逐步完善，传统Python UDF仍无法直接在客户端定义后使用
• 使用Pandas UDF时，需确保客户端与服务器端的Pandas、PyArrow版本兼容，避免出现序列化错误

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