调用PySpark UDF函数报错AttributeError: 'NoneType'对象无'_jvm'属性

错误根因 #

你报错的核心原因是在自定义UDF内部调用了Spark的内置函数unix_timestamp：
Spark的内置函数（pyspark.sql.functions下的所有算子）只能在Driver端运行，用于构造执行计划。UDF的逻辑会被序列化后发送到Executor节点执行，Executor节点上不存在SparkContext（sc）实例，调用unix_timestamp时需要访问sc._jvm，此时sc为None，就触发了'NoneType' object has no attribute '_jvm'的报错。

同时你的代码还存在两个额外问题：

• 两个unix_timestamp相减的结果是秒级数值，你给UDF指定的返回类型是TimestampType，类型完全不匹配
• 你用了Python内置关键字dict作为变量名，会覆盖原生的dict类，存在隐性风险

修复方案 #

方案1：优先使用内置函数实现（性能最优） #

简单的时间差值计算完全不需要自定义UDF，直接用Spark内置算子实现即可，规避UDF的序列化开销，性能比自定义UDF高10倍以上：

import time
import datetime
from pyspark.sql.functions import lit, unix_timestamp, col
from pyspark.sql.types import DecimalType

# 不要用dict作为变量名
dict_data = [{'name': 'Alice', 'age': 1},{'name': 'Again', 'age': 2}]
df = spark.createDataFrame(dict_data)

timestamp1 = datetime.datetime.fromtimestamp(time.time()).strftime('%Y-%m-%d %H:%M:%S')
timestamp2 = datetime.datetime.fromtimestamp(time.time() + 90).strftime('%Y-%m-%d %H:%M:%S')

new_df = (df.withColumn('time1', unix_timestamp(lit(timestamp1),'yyyy-MM-dd HH:mm:ss').cast("timestamp"))
          .withColumn('time2', unix_timestamp(lit(timestamp2),'yyyy-MM-dd HH:mm:ss').cast("timestamp")))

# 直接用内置函数计算时间差，单位为秒
new_df.withColumn("DIFF", (unix_timestamp(col("time2")) - unix_timestamp(col("time1"))).cast(DecimalType(20, 6))).show()

方案2：自定义UDF（仅用于内置函数满足不了需求的场景） #

如果必须使用UDF，UDF内部只能用Python原生的时间处理逻辑，不能调用任何Spark相关的API：

import time
import datetime
from pyspark.sql.functions import lit, unix_timestamp, udf, col
from pyspark.sql.types import LongType, DecimalType

dict_data = [{'name': 'Alice', 'age': 1},{'name': 'Again', 'age': 2}]
df = spark.createDataFrame(dict_data)

timestamp1 = datetime.datetime.fromtimestamp(time.time()).strftime('%Y-%m-%d %H:%M:%S')
timestamp2 = datetime.datetime.fromtimestamp(time.time() + 90).strftime('%Y-%m-%d %H:%M:%S')

# UDF内部使用Python原生datetime方法计算差值
def calc_time(start, end):
    # 传入的start/end已经是Python datetime类型，直接做差返回秒数
    return (end - start).total_seconds()

# 返回值是数值类型，对应LongType
calc_time_udf = udf(calc_time, LongType())

new_df = (df.withColumn('time1', unix_timestamp(lit(timestamp1),'yyyy-MM-dd HH:mm:ss').cast("timestamp"))
          .withColumn('time2', unix_timestamp(lit(timestamp2),'yyyy-MM-dd HH:mm:ss').cast("timestamp")))

new_df.withColumn("DIFF", calc_time_udf(col("time1"), col("time2")).cast(DecimalType(20, 6))).show()

注意事项 #

所有Spark核心对象（SparkSession、SparkContext、pyspark.sql.functions下的算子）都只能在Driver端调用，不要放到UDF、map、foreach这类会发送到Executor执行的逻辑中。

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