Spark-Submit 的酸洗错误“_pickle.PicklingError: args[0] from __newobj__ args has the wrong class”

Question

尝试通过 Spark-Submit 或 Zeppelin 运行某些代码时出现以下错误：“_pickle.PicklingError: args[0] from __ newobj __ args has the wrong class”

我查看了具有相同问题的帖子，但对这个问题没有太多了解。

回溯（包括在下面）指向我使用的 udf 之一：

udf_stop_words = udf(stop_words, ArrayType(StringType()))

def stop_words(words):
    return list(word.lower() for word in words if word.lower() not in  stopwords.words("english"))

函数的输入和输出都是字符串列表。这些是输入中的 3 行：

[Row(split_tokenized_activity_description=['A', 'delightful', '45', '分钟', '瑞典语', '风格', '按摩']), Row(split_tokenized_activity_description=['A', 'more', 'intense', '45', '分钟', '版本', 'of', 'a', '瑞典语', 'style', 'massage']), 行（split_tokenized_activity_description=['A'，'放松'，'45'， '分钟', '瑞典语', '风格', '按摩'])

这是我正在使用的代码的 sn-p。

def special_car(x):
    # remove the special character and replace them with the stop word " " (space)
    return [re.sub('[^A-Za-z0-9]+', ' ', x)]

# Create UDF from function
udf_special_car = udf(special_car, ArrayType(StringType()))

# Function to remove stops words
def stop_words(words):
    return list(word.lower() for word in words if word.lower() not in  stopwords.words("english"))

udf_stop_words = udf(stop_words, ArrayType(StringType()))

# Load in data
df_tags = spark.sql("select * from database")

# Remove special Characters
df1_tags = df_tags.withColumn('tokenized_name', udf_special_car(df_tags.name))
df2_tags = df1_tags.withColumn('tokenized_description', udf_special_car(df1_tags.description))

# Select only relevent columns
df3_tags = df2_tags.select(['tag_id', 'tokenized_name', 'tokenized_description'])

# Tokenize tag_name and tag_desc (Seperate on spaces) (This uses the pyspark.sql.split function)
df4_tags = df3_tags.withColumn('split_tokenized_name', split(df3_tags['tokenized_name'].getItem(0), ' '))
df5_tags = df4_tags.withColumn('split_tokenized_description', split(df3_tags['tokenized_description'].getItem(0), ' '))

# Select only relevent columns
df6_tags = df5_tags.select(['tag_id', 'split_tokenized_name', 'split_tokenized_description'])

# Remove Stop words
df7_tags = df6_tags.withColumn('stop_words_tokenized_name', udf_stop_words(df6_tags.split_tokenized_name))
df8_tags = df7_tags.withColumn('stop_words_tokenized_description', udf_stop_words(df7_tags.split_tokenized_description))

奇怪的是，前两次通过 Zeppelin 运行我的代码时我得到了错误，但在第三次尝试之后，它运行得很好，并且输出是我所期望的。不过，Zeppelin 仅用于测试；我需要让它通过 Spark-Submit 运行。

Traceback（最近一次调用最后一次）：文件 “/tmp/testing_test.py”，第 262 行，在 udf_stop_words = udf(stop_words, ArrayType(StringType())) 文件“/usr/lib/spark/python/lib/pyspark.zip/pyspark/sql/functions.py”，行 1872 年，在 udf 文件中 “/usr/lib/spark/python/lib/pyspark.zip/pyspark/sql/functions.py”，行 1830，在 init 文件中 “/usr/lib/spark/python/lib/pyspark.zip/pyspark/sql/functions.py”，行 1835，在 _create_judf 文件中 “/usr/lib/spark/python/lib/pyspark.zip/pyspark/sql/functions.py”，行 1815，在 _wrap_function 文件中 “/usr/lib/spark/python/lib/pyspark.zip/pyspark/rdd.py”，第 2359 行，在 _prepare_for_python_RDD 文件“/usr/lib/spark/python/lib/pyspark.zip/pyspark/serializers.py”，行 460，在转储文件中 “/usr/lib/spark/python/lib/pyspark.zip/pyspark/cloudpickle.py”，行 703，在转储文件中 “/usr/lib/spark/python/lib/pyspark.zip/pyspark/cloudpickle.py”，行 147、在转储文件“/home/hadoop/anaconda/lib/python3.6/pickle.py”中， 第 409 行，在转储中 self.save(obj) 文件“/home/hadoop/anaconda/lib/python3.6/pickle.py”，第 476 行，保存中 f(self, obj) # 使用显式 self 调用未绑定的方法 File "/home/hadoop/anaconda/lib/python3.6/pickle.py", line 736, in 保存元组 保存（元素）文件“/home/hadoop/anaconda/lib/python3.6/pickle.py”，第 476 行，保存中 f(self, obj) # 使用显式 self 调用未绑定方法 File "/usr/lib/spark/python/lib/pyspark.zip/pyspark/cloudpickle.py", line 248，在 save_function 文件中 “/usr/lib/spark/python/lib/pyspark.zip/pyspark/cloudpickle.py”，行 296，在 save_function_tuple 文件中 “/home/hadoop/anaconda/lib/python3.6/pickle.py”，第 476 行，保存中 f(self, obj) # 使用显式 self 调用未绑定的方法 File "/home/hadoop/anaconda/lib/python3.6/pickle.py", line 821, in 保存字典 self._batch_setitems(obj.items()) 文件“/home/hadoop/anaconda/lib/python3.6/pickle.py”，第 852 行，在 _batch_setitems 保存（v）文件“/home/hadoop/anaconda/lib/python3.6/pickle.py”，第 521 行，保存 self.save_reduce(obj=obj, *rv) 文件“/usr/lib/spark/python/lib/pyspark.zip/pyspark/cloudpickle.py”，行 564，在 save_reduce _pickle.PicklingError: args[0] from newobj args 有错误的类

我已经尝试了几种方法来解决这个问题，但都没有奏效。它们都返回相同的错误。

我尝试将 udf 更改为单行 lambda 函数：

udf(lambda words: list(word.lower() for word in words if word.lower() not in stopwords.words('english')), ArrayType(StringType())).

我已尝试更改 udf 以返回字符串：

udf_stop_words = udf(stop_words, StringType())

并稍微更改 udf 以匹配。

def stop_words(words):
    return str(word.lower() for word in words if word.lower() not in stopwords.words('english'))

我尝试将其定义为 StructType ：

udf_stop_words = udf(stop_words, StructType([StructField("words", ArrayType(StringType()), False)])) 

和

udf_stop_words = udf(stop_words, StructType([StructField("words", StringType(), False)])).

我也尝试了以上的很多组合。

Answer 1

返回类型应该是ArrayType(StringType())。

对此我不确定，但问题可能来自您的节点上没有安装nltk（或者从未在节点上下载corpus stopwords）。由于在 UDF 中调用 stopwords.words("english") 就像在节点上调用它一样，它可能会因为找不到语料库而失败。

由于stopwords.words("english") 只是一个列表，您应该在驱动程序上调用它，然后将其广播到节点：

from nltk.corpus import stopwords
english_stopwords = stopwords.words("english")
sc.broadcast(english_stopwords)
def stop_words(words):
    return list(word.lower() for word in words if word.lower() not in english_stopwords)

from pyspark.sql.types import ArrayType, StringType
import pyspark.sql.functions as psf
udf_stop_words = psf.udf(stop_words, ArrayType(StringType()))

Answer 2

我遇到了类似的问题。在我的情况下，抛出异常是因为我在我的 spark 脚本本身中定义了一个类。它是通过创建一个包含类定义和方法的单独 .py 文件来解决的。然后通过sc.addPyFile(path) 和最后from FileName import * 将该脚本添加到您的脚本中。