Databricks创建Python UDF报错：'spark'未定义

错误原因 #

1. 上下文环境不兼容：Notebook单元格里的spark是Driver端的SparkSession全局对象，可直接调用，但Python UDF运行在Executor节点的分布式环境中，该环境不存在spark全局变量，因此触发NameError。
2. 逻辑设计反模式：在UDF内部执行spark.sql属于严重低效写法——每调用一次UDF就会触发独立的SQL查询，数据量较大时会引发性能雪崩，完全违背Spark分布式计算的设计初衷。

正确实现方式 #

方案一：原生SQL直接实现（最优解） #

无需自定义UDF，直接用SQL完成聚合查询：

SELECT SUM(count_of_runs) AS UDF_Result
FROM mysqltableinDBX
WHERE serial = 71605;

方案二：复用逻辑的SQL函数实现 #

如果需要复用该逻辑，可创建SQL自定义函数（而非Python UDF）：

CREATE OR REPLACE FUNCTION get_total_runs(serial_input INT)
RETURNS INT
LANGUAGE SQL
AS $$
  SELECT SUM(count_of_runs)
  FROM mysqltableinDBX
  WHERE serial = serial_input
$$;

-- 调用方式
SELECT get_total_runs(71605) AS UDF_Result;

方案三：Python逻辑的合规实现（仅适用于复杂场景） #

若必须用Python处理复杂逻辑，绝对不能在UDF内触发Spark查询，需提前加载数据并通过广播变量传递：

from pyspark.sql import functions as F
from pyspark.sql.types import IntegerType

# 预聚合目标数据并广播到Executor
agg_df = spark.sql("SELECT serial, SUM(count_of_runs) AS total_runs FROM mysqltableinDBX GROUP BY serial")
broadcast_data = F.broadcast(agg_df)

# 定义UDF（实际场景优先用DataFrame关联，此示例仅作演示）
@F.udf(IntegerType())
def myUDF(serial_input):
    # 从广播数据中匹配结果
    result = [row.total_runs for row in broadcast_data.collect() if row.serial == serial_input]
    return result[0] if result else 0

# 使用UDF（更推荐直接用DataFrame join替代）
spark.sql("SELECT 71605 AS serial").withColumn("UDF_Result", myUDF(F.col("serial"))).show()

核心原则：优先使用原生SQL或Spark内置函数实现需求，避免在UDF中执行Spark查询类操作。

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