PySpark中UDF使用round函数报错原因咨询

原始预处理代码 #

def data_preparation(df):

    unlist = udf(lambda x: round(float(list(x)[0]),3), FloatType())
    # Iterating over columns to be scaled
    for i in ["event"]:
        # VectorAssembler Transformation - Converting column to vector type
        assembler = VectorAssembler(inputCols=[i],outputCol=i+"_Vect")

        # MinMaxScaler Transformation
        scaler = MinMaxScaler(inputCol=i+"_Vect", outputCol=i+"_Scaled")

        # Pipeline of VectorAssembler and MinMaxScaler
        pipeline = Pipeline(stages=[assembler, scaler])

        # Fitting pipeline on dataframe
        df = pipeline.fit(df).transform(df).withColumn(i+"_Scaled",unlist(i+"_Scaled")).drop(i+"_Vect")

    return df

报错信息 #

PythonException: 
  An exception was thrown from the Python worker. Please see the stack trace below.
Traceback (most recent call last):
  File "/usr/local/lib/python3.6/dist-packages/pyspark/python/lib/pyspark.zip/pyspark/worker.py", line 604, in main
    process()
  File "/usr/local/lib/python3.6/dist-packages/pyspark/python/lib/pyspark.zip/pysparkx/worker.py", line 596, in process
    serializer.dump_stream(out_iter, outfile)
  File "/usr/local/lib/python3.6/dist-packages/pyspark/python/lib/pyspark.zip/pyspark/serializers.py", line 211, in dump_stream
    self.serializer.dump_stream(self._batched(iterator), stream)
  File "/usr/local/lib/python3.6/dist-packages/pyspark/python/lib/pyspark.zip/pyspark/serializers.py", line 132, in dump_stream
    for obj in iterator:
  File "/usr/local/lib/python3.6/dist-packages/pyspark/python/lib/pyspark.zip/pyspark/serializers.py", line 200, in _batched
    for item in iterator:
  File "/usr/local/lib/python3.6/dist-packages/pyspark/python/lib/pyspark.zip/pyspark/worker.py", line 450, in mapper
    result = tuple(f(*[a[o] for o in arg_offsets]) for (arg_offsets, f) in udfs)
  File "/usr/local/lib/python3.6/dist-packages/pyspark/python/lib/pyspark.zip/pyspark/worker.py", line 450, in <genexpr>
    result = tuple(f(*[a[o] for o in arg_offsets]) for (arg_offsets, f) in udfs)
  File "/usr/local/lib/python3.6/dist-packages/pyspark/python/lib/pyspark.zip/pyspark/worker.py", line 85, in <lambda>
    return lambda *a: f(*a)
  File "/usr/local/lib/python3.6/dist-packages/pyspark/python/lib/pyspark.zip/pyspark/util.py", line 73, in wrapper
    return f(*args, **kwargs)
  File "<ipython-input-85-a4273b6bc9ab>", line 17, in <lambda>
  File "/usr/local/lib/python3.6/dist-packages/pyspark/python/lib/pyspark.zip/pyspark/sql/functions.py", line 1234, in round
    return Column(sc._jvm.functions.round(_to_java_column(col), scale))
AttributeError: 'NoneType' object has no attribute '_jvm'

错误原因分析 #

错误的核心是UDF中误用了PySpark SQL的round函数，而非Python内置的round：

• PySpark的pyspark.sql.functions.round是面向Column对象的分布式计算API，它依赖SparkContext（sc）的_jvm连接来调用Java端的计算逻辑。
• UDF运行在独立的Python worker进程中，这些进程没有初始化SparkContext，因此sc为None，调用sc._jvm就会触发AttributeError。
• 出现这个问题的前提是你代码中导入了from pyspark.sql.functions import round，导致UDF内的round指向了Spark的函数，覆盖了Python原生的round。

解决方案 #

方案1：明确调用Python内置round函数 #

导入builtins模块，直接指定使用Python原生的round：

import builtins
from pyspark.sql.functions import udf
from pyspark.sql.types import FloatType
from pyspark.ml.feature import VectorAssembler, MinMaxScaler
from pyspark.ml import Pipeline

def data_preparation(df):
    unlist = udf(lambda x: builtins.round(float(list(x)[0]), 3), FloatType())
    for i in ["event"]:
        assembler = VectorAssembler(inputCols=[i], outputCol=i+"_Vect")
        scaler = MinMaxScaler(inputCol=i+"_Vect", outputCol=i+"_Scaled")
        pipeline = Pipeline(stages=[assembler, scaler])
        df = pipeline.fit(df).transform(df).withColumn(i+"_Scaled", unlist(i+"_Scaled")).drop(i+"_Vect")
    return df

方案2：修改导入语句避免覆盖 #

将Spark的round函数重命名导入，保留Python原生round的使用：

from pyspark.sql.functions import udf, round as spark_round
from pyspark.sql.types import FloatType
from pyspark.ml.feature import VectorAssembler, MinMaxScaler
from pyspark.ml import Pipeline

def data_preparation(df):
    unlist = udf(lambda x: round(float(list(x)[0]), 3), FloatType())
    for i in ["event"]:
        assembler = VectorAssembler(inputCols=[i], outputCol=i+"_Vect")
        scaler = MinMaxScaler(inputCol=i+"_Vect", outputCol=i+"_Scaled")
        pipeline = Pipeline(stages=[assembler, scaler])
        df = pipeline.fit(df).transform(df).withColumn(i+"_Scaled", unlist(i+"_Scaled")).drop(i+"_Vect")
    return df

方案3：弃用UDF，使用Spark原生向量操作（推荐） #

无需自定义UDF，直接用Spark内置的element_at函数提取向量元素，搭配Spark的round函数，性能更优：

from pyspark.sql.functions import col, element_at, round
from pyspark.ml.feature import VectorAssembler, MinMaxScaler
from pyspark.ml import Pipeline

def data_preparation(df):
    for i in ["event"]:
        assembler = VectorAssembler(inputCols=[i], outputCol=i+"_Vect")
        scaler = MinMaxScaler(inputCol=i+"_Vect", outputCol=i+"_Scaled")
        pipeline = Pipeline(stages=[assembler, scaler])
        df = pipeline.fit(df).transform(df)
        # 提取向量第一个元素并四舍五入
        df = df.withColumn(i+"_Scaled", round(element_at(col(i+"_Scaled"), 1), 3)).drop(i+"_Vect")
    return df

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