mllib NaiveBayes 中的类数量是否有限制？调用 model.save() 时出错

Question

我正在尝试训练一个模型来预测文本输入数据的类别。当类的数量超过一定数量时，我在词袋上使用pyspark.ml.classification.NaiveBayes 分类器遇到了似乎是数值不稳定性的问题。

在我的真实项目中，我有大约 10 亿条记录和大约 50 个类。我能够训练我的模型并做出预测，但是当我尝试使用 model.save() 保存它时出现错误。从操作上讲，这很烦人，因为我每次都必须从头开始重新训练我的模型。

在尝试调试时，我将数据缩减到约 10k 行，并在尝试保存时遇到了同样的问题。但是，如果我减少类标签的数量，保存效果很好。

这让我相信标签的数量是有限制的。我无法重现我的确切问题，但下面的代码是相关的。如果我将 num_labels 设置为大于 31 的任何值，model.fit() 会引发错误。

我的问题：

1. NaiveBayes 的 mllib 实现中的类数量是否有限制？
2. 如果我可以成功地使用模型进行预测，我无法保存模型的原因有哪些？
3. 如果确实有限制，是否可以将我的数据分成更小的类别、训练单独的模型并合并？

完整的工作示例

创建一些虚拟数据。

我将使用nltk.corpus.comparitive_sentences 和nltk.corpus.sentence_polarity。请记住，这只是一个带有无意义数据的说明性示例 - 我不关心拟合模型的性能。

import pandas as pd
from pyspark.sql.types import StringType

# create some dummy data
from nltk.corpus import comparative_sentences, sentence_polarity
df = pd.DataFrame(
    {
        'sentence': [" ".join(s) for s in cs.sents() + sp.sents()]
    }
)

# assign a 'category' to each row
num_labels = 31  # seems to be the upper limit
df['category'] = (df.index%num_labels).astype(str)

# make it into a spark dataframe
spark_df = sqlCtx.createDataFrame(df)

数据准备管道

from pyspark.ml.feature import NGram, Tokenizer, StopWordsRemover
from pyspark.ml.feature import HashingTF, IDF, StringIndexer, VectorAssembler
from pyspark.ml import Pipeline
from pyspark.ml.linalg import Vector

indexer = StringIndexer(inputCol='category', outputCol='label')
tokenizer = Tokenizer(inputCol="sentence", outputCol="sentence_tokens")
remove_stop_words = StopWordsRemover(inputCol="sentence_tokens", outputCol="filtered")
unigrammer = NGram(n=1, inputCol="filtered", outputCol="tokens") 
hashingTF = HashingTF(inputCol="tokens", outputCol="hashed_tokens")
idf = IDF(inputCol="hashed_tokens", outputCol="tf_idf_tokens")

clean_up = VectorAssembler(inputCols=['tf_idf_tokens'], outputCol='features')

data_prep_pipe = Pipeline(
    stages=[indexer, tokenizer, remove_stop_words, unigrammer, hashingTF, idf, clean_up]
)
transformed = data_prep_pipe.fit(spark_df).transform(spark_df)
clean_data = transformed.select(['label','features'])

训练模型

from pyspark.ml.classification import NaiveBayes
nb = NaiveBayes()
(training,testing) = clean_data.randomSplit([0.7,0.3], seed=12345)
model = nb.fit(training)
test_results = model.transform(testing)

评估模型

from pyspark.ml.evaluation import MulticlassClassificationEvaluator
acc_eval = MulticlassClassificationEvaluator()
acc = acc_eval.evaluate(test_results)
print("Accuracy of model at predicting label was: {}".format(acc))

在我的机器上，打印：

Accuracy of model at predicting label was: 0.0305764788269

错误信息

如果我将 num_labels 更改为 32 或更高，这是我在调用 model.fit() 时遇到的错误：

Py4JJavaError：调用 o1336.fit 时出错。 ： org.apache.spark.SparkException：作业因阶段失败而中止： 阶段 86.0 中的任务 0 失败 4 次，最近一次失败：丢失任务 0.3 阶段 86.0（TID 1984，someserver.somecompany.net，执行程序 22）：org.apache.spark.SparkException：Kryo 序列化失败：缓冲区 溢出。可用：7，必需：8 序列化跟踪：值 （org.apache.spark.ml.linalg.DenseVector）。为了避免这种情况，增加 spark.kryoserializer.buffer.max 值。 ... ... 等等等等更多的java东西，永远持续下去

注意事项

• 在这个例子中，如果我为二元组添加一个特征，如果num_labels > 15 就会发生错误。我想知道这也是 1 小于 2 的幂是否是巧合。
• 在我的实际项目中，尝试调用model.theta 时也会出错。 （我不认为错误本身是有意义的——它们只是从 java/scala 方法传回的异常。）

Answer 1

硬性限制：

特征数 * 类数必须小于 Integer.MAX_VALUE (2³¹ - 1)。你离这些价值还差得很远。

软限制：

Theta 矩阵（条件概率）的大小为特征数 * 类数。 Theta 存储在本地驱动程序（作为模型的一部分）并序列化并发送给工作人员。这意味着所有机器至少需要足够的内存来序列化或反序列化并存储结果。

由于您使用 HashingTF.numFeatures (2²⁰) 的默认设置，因此每个额外的类都会增加 262144 - 数量不多，但很快就会增加。根据您发布的部分回溯，看起来失败的组件是 Kryo 序列化程序。同样的回溯也提出了解决方案，即增加spark.kryoserializer.buffer.max。

您也可以通过设置尝试使用标准 Java 序列化：

 spark.serializer org.apache.spark.serializer.JavaSerializer 

由于您将 PySpark 与 pyspark.ml 和 pyspark.sql 一起使用，因此在没有显着性能损失的情况下可能是可以接受的。

除了配置，我将专注于特征工程组件。使用二进制 CountVetorizer（请参阅下面关于 HashingTF 的注释）和 ChiSqSelector 可能会提供一种方法来提高可解释性并有效减少功能数量。您还可以考虑更复杂的方法（确定特征重要性并仅在数据子集上应用朴素贝叶斯，更高级的文本处理，如词形还原/词干提取，或使用自动编码器的某些变体来获得更紧凑的向量表示）。

注意事项：

• 请记住，跨国朴素贝叶斯只考虑二元特征。 NaiveBayes 将在内部处理此问题，但为了清楚起见，我仍然建议使用 setBinary。
• 可以说HashingTF 在这里毫无用处。撇开哈希冲突不谈，高度稀疏的特征和本质上无意义的特征，使其成为NaiveBayes 的预处理步骤的糟糕选择。