执行字符串匹配时遇AttributeError：DataFrame无withColumn属性求助

1. df_merged是pandas DataFrame：你用了ps.merge（ps应为pandas的别名），得到的结果并非Spark DataFrame，而withColumn是Spark DataFrame的专属方法，因此触发AttributeError。
2. UDF返回类型错误：fuzz.token_sort_ratio返回的是0-100的整数相似度得分，你定义的StringType类型不匹配，后续会引发额外报错。

方案一：全程使用Spark处理（推荐） #

将pandas的合并操作替换为Spark的join，保持数据在Spark生态内处理：

from pyspark.sql import functions as f
from fuzzywuzzy import fuzz
from pyspark.sql.types import IntegerType
from pyspark.sql import SparkSession

def matchstring(s1, s2):
    # 处理空值，避免fuzz计算报错
    if not s1 or not s2:
        return 0
    return fuzz.token_sort_ratio(s1, s2)

# 修正UDF返回类型为IntegerType
MatchUDF = f.udf(matchstring, IntegerType())

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

# 若原始数据是pandas DataFrame，先转为Spark DataFrame
# df_Sale_KR_spark = spark.createDataFrame(df_Sale_KR)
# df_Dist_Mast_spark = spark.createDataFrame(df_Dist_Mast)

# 用Spark join替代pandas merge
df_merged = df_Sale_KR.join(df_Dist_Mast, on='Distributor_ID', how='left')

# 正常调用withColumn生成新列
df_similarity_score = df_merged.withColumn("similarity_score", MatchUDF(f.col("source"), f.col("target")))
df_similarity_score.show()

方案二：继续用pandas处理 #

如果数据量较小，可保留pandas DataFrame，直接用apply计算相似度：

import pandas as pd
from fuzzywuzzy import fuzz

def matchstring(s1, s2):
    if not s1 or not s2:
        return 0
    return fuzz.token_sort_ratio(s1, s2)

# 原有pandas合并逻辑
df_merged = pd.merge(df_Sale_KR, df_Dist_Mast, on='Distributor_ID', how='left')

# 生成相似度得分列
df_merged['similarity_score'] = df_merged.apply(lambda row: matchstring(row['source'], row['target']), axis=1)

# 查看结果
print(df_merged.head())

注：大数据量场景优先选方案一，Spark的分布式处理效率更高；小数据量用方案二更轻便。

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