在PySpark DataFrame列中应用UDF时遭遇序列化错误

错误原因 #

词形还原工具（比如NLTK的WordNetLemmatizer、Spacy的NLP模型）内部会维护_thread.RLock这类用于线程安全的对象，这类对象属于Python底层线程资源，无法被序列化。

PySpark执行UDF时，会把UDF函数及其依赖的所有对象序列化后分发到集群各个工作节点执行。如果在UDF外部提前初始化了词形还原工具实例，这个实例会携带不可序列化的线程锁对象，导致序列化失败；而预处理、去停用词的函数通常只是字符串操作或调用无状态工具，没有持有这类不可序列化资源，所以能正常运行。

解决办法 #

1. 在UDF内部延迟初始化词形还原工具 #

不要在驱动端提前创建词形还原实例，而是在UDF函数内部首次调用时初始化，让每个工作节点的进程独立创建实例，避免序列化问题。以NLTK的WordNetLemmatizer为例：

from pyspark.sql.functions import udf
from nltk.stem import WordNetLemmatizer

def lemmatize_text(text):
    # 函数内部初始化lemmatizer，每个节点进程独立创建
    lemmatizer = WordNetLemmatizer()
    words = text.split()
    return [lemmatizer.lemmatize(word) for word in words]

lemmatize_udf = udf(lemmatize_text)
df = df.withColumn("lemmatized_words", lemmatize_udf("clean_text"))

注意：需确保集群所有节点已下载NLTK所需语料（如WordNet），可在集群初始化脚本中执行nltk.download('wordnet')。

2. 使用PySpark原生MLlib工具替代自定义UDF #

PySpark MLlib提供了分布式原生的词形/词干处理工具，无需手动编写UDF，从根源避免序列化问题。示例：

from pyspark.ml.feature import Lemmatizer, Tokenizer
from pyspark.ml import Pipeline

# 先分词，再词形还原
tokenizer = Tokenizer(inputCol="text", outputCol="words")
lemmatizer = Lemmatizer(inputCol="words", outputCol="lemmatized_words")

# 构建Pipeline执行
pipeline = Pipeline(stages=[tokenizer, lemmatizer])
result_df = pipeline.fit(df).transform(df)

如果版本不支持Lemmatizer，也可以用SnowballStemmer做词干提取（需求允许的情况下）。

3. 针对Spacy等重型NLP库的优化 #

如果使用Spacy，同样在UDF内部延迟加载模型，同时确保集群节点已安装对应模型：

import spacy
from pyspark.sql.functions import udf

def spacy_lemmatize(text):
    # 每个节点进程独立加载模型
    nlp = spacy.load("en_core_web_sm")
    doc = nlp(text)
    return [token.lemma_ for token in doc]

spacy_lemmatize_udf = udf(spacy_lemmatize)
df = df.withColumn("lemmatized_text", spacy_lemmatize_udf("text"))

若模型体积较大，可将模型文件上传到集群共享存储，让节点从共享路径加载，避免重复下载。

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