PySpark应用自定义UDF执行失败问题求助

问题重现 #

代码示例：

multiplier = udf(lambda x: float(x) * 100.0, FloatType())
df = df.select(multiplier(df['value']).alias('value_percent'))

抛出错误：

Lost task 0.0 in stage 1.0 (TID 1) (127.0.0.1 executor driver):
org.apache.spark.SparkException: Python worker exited unexpectedly
....
java.io.EOFException

环境信息：Spark 3.5.3，Python 3.11.9，Windows系统，相同代码、数据集在Jupyter Notebook可正常运行。

解决方案 #

• 指定Python解释器路径
  Windows环境下Spark可能无法自动识别正确的Python版本，需手动绑定解释器路径。可在运行脚本前通过命令行设置：

  set PYSPARK_PYTHON=C:\Python311\python.exe
  set PYSPARK_DRIVER_PYTHON=C:\Python311\python.exe
  

  或在Python脚本开头添加：

  import os
  os.environ['PYSPARK_PYTHON'] = 'C:\\Python311\\python.exe'
  os.environ['PYSPARK_DRIVER_PYTHON'] = 'C:\\Python311\\python.exe'
  

• 替换lambda UDF为具名函数
  lambda表达式在Windows的Spark进程通信中可能存在兼容性问题，改用具名函数定义UDF：

  from pyspark.sql.functions import udf
  from pyspark.sql.types import FloatType
  
  def multiplier(x):
      return float(x) * 100.0
  
  multiplier_udf = udf(multiplier, FloatType())
  df = df.select(multiplier_udf(df['value']).alias('value_percent'))
  

• 添加异常处理兼容异常数据
  数据集可能存在空值或非数值类型数据，在UDF中增加异常捕获逻辑：

  def multiplier(x):
      if x is None:
          return None
      try:
          return float(x) * 100.0
      except (ValueError, TypeError):
          return None
  

• 调整Python Worker内存
  Windows下默认Worker内存可能不足，初始化SparkSession时增加内存配置：

  from pyspark.sql import SparkSession
  
  spark = SparkSession.builder \
      .appName("UDFDemo") \
      .config("spark.python.worker.memory", "1g") \
      .getOrCreate()
  

• 排查防火墙/杀毒软件拦截
  Windows防火墙或杀毒软件可能阻断Spark Worker进程的通信，临时关闭相关软件测试是否解决问题。

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