apache-spark - 文本分类——如何处理

Question

我会尝试描述我的想法。

MS SQL 数据库中存储有一段文本内容。内容每天都会以流的形式出现。有些人每天都会浏览内容，如果内容符合特定标准，则将其标记为已验证。只有一类。它要么“有效”，要么无效。

我想要的是根据已经验证的内容创建一个模型，保存它并使用该模型来“预验证”或标记新的传入内容。偶尔也会根据新验证的内容更新模型。希望我能清楚地解释自己。

我正在考虑使用 Spark 流根据创建的模型进行数据分类。以及朴素贝叶斯算法。但是您将如何创建、更新和存储模型呢？有大约 200K+ 不同长度的经过验证的结果(文本)。模型需要这么多吗？以及如何在 Spark Streaming 中使用该模型。

提前致谢。

Answer 1

哇这个问题非常广泛，与机器学习比与Apache Spark更相关，但是我会尝试给您一些提示或遵循的步骤(我不会为您做这项工作)。

1. 导入您需要的所有库

   from pyspark.mllib.classification import LogisticRegressionWithSGD, LogisticRegressionModel
   from pyspark.mllib.linalg import SparseVector
   from pyspark.mllib.regression import LabeledPoint
   import re
   

2. 将数据加载到 RDD

   msgs = [("I love Star Wars but I can't watch it today", 1.0),
           ("I don't love Star Wars and people want to watch it today", 0.0),
           ("I dislike not being able to watch Star Wars", 1.0),
           ("People who love Star Wars are my friends", 1.0),
           ("I preffer to watch Star Wars on Netflix", 0.0),
           ("George Lucas shouldn't have sold the franchise", 1.0),
           ("Disney makes better movies than everyone else", 0.0)]
   
   rdd = sc.parallelize(msgs)
   

3. 标记化您的数据(如果您使用ML可能会更容易)和

   rdd = rdd.map(lambda (text, label): ([w.lower() for w in re.split(" +", text)], label))
   

4. 删除所有不必要的单词(广泛称为停止词)和符号，例如,.&

   commons = ["and", "but", "to"]
   rdd = rdd.map(lambda (tokens, label): (filter(lambda token: token not in commons, tokens), label))
   

5. 使用所有数据集中中的所有不同单词创建一个字典，听起来很大，但它们并没有您期望的那么多，我敢打赌它们将适合您的主节点(但是还有其他方法可以解决此问题，但为了简单起见，我将保留这种方式)。

   # finds different words
   words = rdd.flatMap(lambda (tokens, label): tokens).distinct().collect()
   diffwords = len(words)
   

6. 将您的功能转换为 DenseVector或SparseVector ，我显然会推荐第二种方式，因为通常 SparseVector 需要更少的空间来表示，但这取决于数据。请注意，还有更好的替代方法，例如散列，但我试图忠于我的详细方法。之后将元组转换为 LabeledPoint

   def sparsify(length, tokens):
       indices = [words.index(t) for t in set(tokens)]
       quantities = [tokens.count(words[i]) for i in indices]
   
       return SparseVector(length, [(indices[i], quantities[i]) for i in xrange(len(indices))])
   
   rdd = rdd.map(lambda (tokens, label): LabeledPoint(label, sparsify(diffwords, tokens)))
   

7. 适合您最喜欢的型号，在本例中我使用 LogisticRegressionWithSGD是别有用心的。

   lrm = LogisticRegressionWithSGD.train(rdd)
   

8. Save你的模型。

   lrm.save(sc, "mylovelymodel.model")
   

9. Load你的LogisticRegressionModel在另一个应用程序中。

   lrm = LogisticRegressionModel.load(sc, "mylovelymodel.model")
   

10. Predict类别。

    lrm.predict(SparseVector(37,[2,4,5,13,15,19,23,26,27,29],[1.0,1.0,1.0,1.0,1.0,1.0,1.0,1.0,1.0,1.0]))
    # outputs 0
    

请注意，我没有评估模型的准确性，但它看起来非常不是吗？