Spark UDF报错修复与代码优化：补全medicine_name空值

让我来帮你修复这个Spark代码的问题，顺便优化一下实现逻辑——你的报错是因为几个违反Spark执行模型的问题，我会一步步拆解并给出完整的解决方案：

先说说你现有代码的核心问题 #

1. 未注册UDF就直接在SQL中调用：editDistanceUdf没有注册为Spark SQL可识别的函数，直接写在spark.sql()里会导致Spark无法解析。
2. UDF内部执行Spark SQL操作：getSimilarity函数在UDF里调用spark.sql()是完全错误的——UDF运行在Executor节点，而SparkSession是Driver端的专属对象，Executor无法直接执行Spark SQL操作，这就是你看到AttributeError: 'function' object has no attribute '_get_object_id'的原因。
3. 语法错误：when函数的括号不匹配，还有df.medicine_name.=="null"多了个多余的点，导致语法解析失败。
4. 自定义逻辑冗余：自己实现的getMax完全可以用Python内置的max()函数替代，没必要重复造轮子。

修复并优化后的完整代码 #

第一步：导入依赖库

from pyspark.sql import SparkSession
from pyspark.sql.functions import udf, when, col, broadcast
from pyspark.sql.types import StringType, FloatType
# 推荐使用成熟的python-Levenshtein库，比自己实现更高效准确
from Levenshtein import ratio

第二步：初始化SparkSession（如果还没初始化）

spark = SparkSession.builder.appName("MedicineNameCompletion").getOrCreate()

第三步：预处理非空药品名并广播

把非空的药品名列表广播到所有Executor，避免每个Task重复读取数据，大幅提升性能：

# 获取所有非空的medicine_name并去重
distinct_med_names = [
    row.medicine_name 
    for row in df.filter(col("medicine_name").isNotNull())
                 .select("medicine_name")
                 .distinct()
                 .collect()
]
# 广播变量，分发到所有Executor节点
broadcast_med_names = spark.sparkContext.broadcast(distinct_med_names)

第四步：定义匹配逻辑的UDF

在UDF内部使用广播变量的本地数据进行匹配，完全符合Spark的执行模型：

def find_matching_medicine(original_name):
    # 处理原始名称为空的情况
    if not original_name:
        return None
    
    med_names = broadcast_med_names.value
    max_similarity = 0.0
    best_match = None
    
    for med_name in med_names:
        if not med_name:
            continue
        # 忽略大小写计算相似度，ratio返回0-1的数值，值越高越相似
        similarity = ratio(original_name.lower(), med_name.lower())
        # 只保留相似度>=0.85的最优匹配
        if similarity >= 0.85 and similarity > max_similarity:
            max_similarity = similarity
            best_match = med_name
    
    return best_match

# 注册UDF
find_match_udf = udf(find_matching_medicine, StringType())

第五步：应用UDF完成数据补全

# 补全逻辑：当medicine_name为空时调用匹配UDF，否则保留原数据
result_df = df.withColumn(
    "medicine_name",
    when(
        col("medicine_name").isNull(),
        find_match_udf(col("original_name"))
    ).otherwise(col("medicine_name"))
)

# 查看最终结果
result_df.show(truncate=False)

额外说明：如果无法安装python-Levenshtein库 #

如果你的Spark集群无法安装第三方库，可以自己实现Levenshtein相似度计算，替换掉上面的ratio方法：

def levenshtein_ratio(s1, s2):
    s1 = s1.lower()
    s2 = s2.lower()
    m, n = len(s1), len(s2)
    
    # 初始化动态规划矩阵
    dp = [[0]*(n+1) for _ in range(m+1)]
    for i in range(m+1):
        dp[i][0] = i
    for j in range(n+1):
        dp[0][j] = j
    
    # 填充矩阵计算编辑距离
    for i in range(1, m+1):
        for j in range(1, n+1):
            if s1[i-1] == s2[j-1]:
                dp[i][j] = dp[i-1][j-1]
            else:
                dp[i][j] = min(dp[i-1][j], dp[i][j-1], dp[i-1][j-1]) + 1
    
    max_len = max(m, n)
    return (max_len - dp[m][n]) / max_len if max_len != 0 else 1.0

然后在find_matching_medicine函数里用levenshtein_ratio(original_name, med_name)替代ratio(...)即可。

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