问题陈述和约束

1. 我们试图解决的问题是什么？

   我们想要构建一个应用程序来获取电影评论的文本并生成分数。我们将使用它来汇总评论，因此该应用程序将作为批处理运行。我们将使用一个显示器向用户展示这一点，该显示器显示电影收到的正面或负面程度。我们不会担心演示文稿的其他方面。我们将假设显示将嵌入到其他内容中。

2. 有哪些限制条件？

   这是我们的约束：

   • 我们假设我们正在处理英语评论。
   • 我们对程序的速度没有太大的限制，因为这是一个批量离线过程。我们希望在不到 1 分钟的时间内返回 95% 电影的总分。
   • 我们想确保我们的模型在这个任务上表现良好，所以我们将使用一个众所周知的数据集。我们将使用基于 IMDb 用户评论的大型电影评论数据集。
   • 我们将假设该程序的输入是评论的JavaScript Object Notation (JSON)文件。输出将是一个分数。
   • 该模型在新数据上的 F1 分数必须至少为 0.7。

   在我们查看数据之前设置所需的度量阈值似乎不合理，但这种情况很常见。与利益相关者进行谈判很重要。如果设置了任意阈值，但实验表明这是不现实的，那么数据科学家应该能够向利益相关者解释为什么问题比预期的更困难。

   在您处理项目时，此列表可能会发生变化。越早发现遗漏的约束越好。如果您在部署之前发现了一个约束，那么修复它可能会非常昂贵。这就是为什么我们要在开发过程中与利益相关者进行迭代。

   现在我们已经列出了我们的约束，让我们讨论如何构建我们的应用程序。

3. 我们如何解决约束问题？

   第一个限制，即评论是英文的，实际上使我们的任务更容易。第二个约束，关于计算总分需要多长时间，控制我们可以构建的模型的复杂程度，但它是一个轻约束。我们有 IMDb 数据集。当我们构建程序时，我们将从 JSON 加载。但是，我们的建模代码不需要遵循这样的约束。

计划项目

为了计划项目，让我们定义我们的验收标准是什么。产品所有者通常会通过合并利益相关者的请求来定义这些。在本章中，您既是产品所有者又是开发人员。

我们需要一个执行以下操作的脚本：

• 获取带有 JSON 对象评论的文件，每行一个
• 根据模型的输出返回分布信息
  • Mean
  • Standard deviation
  • Quartiles
  • Min
  • Max

我们将使用 Spark NLP 处理数据并使用 Spark MLlib 模型来预测情绪。

现在我们有了这些高级验收标准，让我们看一下数据。首先，我们将数据加载到DataFrames 中并添加标签列。

import sparknlp

from pyspark.ml import Pipeline
from pyspark.sql import SparkSession
from pyspark.sql.functions import lit

import sparknlp
from sparknlp import DocumentAssembler, Finisher
from sparknlp.annotator import *

spark = sparknlp.start()

pos_train = spark.sparkContext.wholeTextFiles(
    'aclImdb_v1/aclImdb/train/pos/')
neg_train = spark.sparkContext.wholeTextFiles(
    'aclImdb_v1/aclImdb/train/neg/')
pos_test = spark.sparkContext.wholeTextFiles(
    'aclImdb_v1/aclImdb/test/pos/')
neg_test = spark.sparkContext.wholeTextFiles(
    'aclImdb_v1/aclImdb/test/neg/')

pos_train = spark.createDataFrame(pos_train, ['path', 'text'])
pos_train = pos_train.distribution(100)
pos_train = pos_train.withColumn('label', lit(1)).persist()
neg_train = spark.createDataFrame(neg_train, ['path', 'text'])
neg_train = neg_train.repartition(100)
neg_train = neg_train.withColumn('label', lit(0)).persist()
pos_test = spark.createDataFrame(pos_test, ['path', 'text'])
pos_test = pos_test.repartition(100)
pos_test = pos_test.withColumn('label', lit(1)).persist()
neg_test = spark.createDataFrame(neg_test, ['path', 'text'])
neg_test = neg_test.repartition(100)
neg_test = neg_test.withColumn('label', lit(0)).persist()

让我们看一个正面的例子。

print(pos_train.first()['text'])

I laughed a lot while watching this. It's an amusing short with a 
fun musical act and a lot of wackiness. The characters are simple, 
but their simplicity adds to the humor stylization. The dialog is 
funny and often unexpected, and from the first line to the last 
everything just seems to flow wonderfully. There's Max, who has 
apparently led a horrible life. And there's Edward, who isn't sure 
what life he wants to lead. My favorite character was Tom, Edward's 
insane boss. Tom has a short role but a memorable one. Highly 
recommended for anyone who likes silly humor. And you can find it 
online now, which is a bonus! I am a fan of all of Jason's cartoons 
and can't wait to see what he comes out with next.

这似乎是一个明显的正面评价。我们可以识别出一些看起来像是一个好信号的词，比如“最好的”。

现在，让我们看一个否定的例子。

print(neg_train.first()['text'])

I sat glued to the screen, riveted, yawning, yet keeping an 
attentive eye. I waited for the next awful special effect, or the 
next ridiculously clichéd plot item to show up full force, so I 
could learn how not to make a movie.<br /><br />It seems when they 
set out to make this movie, the crew watched every single other 
action/science-fiction/shoot-em-up/good vs. evil movie ever made, 
and saw cool things and said: "Hey, we can do that." For example, 
the only car parked within a mile on what seems like a one way road 
with a shoulder not meant for parking, is the one car the 
protagonist, an attractive brunette born of bile, is thrown on to. 
The car blows to pieces before she even lands on it. The special 
effects were quite obviously my biggest beef with this movie. But 
what really put it in my bad books was the implausibility, and lack 
of reason for so many elements! For example, the antagonist, a 
flying demon with the ability to inflict harm in bizarre ways, 
happens upon a lone army truck transporting an important VIP. 
Nameless security guys with guns get out of the truck, you know 
they are already dead. Then the guy protecting the VIP says "Under 
no circumstances do you leave this truck, do you understand me?" He 
gets out to find the beast that killed his 3 buddies, he gets 
whacked in an almost comically cliché fashion. Then for no apparent 
reason, defying logic, convention, and common sense, the dumb ass 
VIP GETS OUT OF THE TRUCK!!! A lot of what happened along the 
course of the movie didn't make sense. Transparent acting distanced 
me from the movie, as well as bad camera-work, and things that just 
make you go: "Wow, that's incredibly cheesy." Shiri Appleby saved 
the movie from a 1, because she gave the movie the one element that 
always makes viewers enjoy the experience, sex appeal.

这是负面评论的一个明显例子。我们在这里看到很多看起来像是负面情绪的可靠指标的词，比如““awful” and “cheesy.”

请注意，我们将要删除一些 HTML 工件。

现在，让我们看一下整个语料库。

print('pos_train size', pos_train.count())
print('neg_train size', neg_train.count())
print('pos_test size', pos_test.count())
print('neg_test size', neg_test.count())

pos_train size 12500
neg_train size 12500
pos_test size 12500
neg_test size 12500

所以我们有 50,000 个文档。在这种情况下，正负之间的均匀分布是人为的。让我们看一下文本的长度，如表 12-1所示。

pos_train.selectExpr('length(text) AS text_len')\
    .toPandas().describe()

字符长度似乎有很多变化。这可能是一个有用的功能。文本长度可能看起来非常低，但它通常是关于文本的有用信息。我们可能会发现，较长的评论更可能因咆哮而变得负面。在这种情况下，如果我们有审稿人 ID 会更有用，这样我们就可以了解审稿人的正常情况；唉，这不在数据中。

现在我们已经简要了解了数据，让我们开始设计我们的解决方案。

设计解决方案

首先，让我们将项目分为两个阶段。

1. 训练和测量模型

   建模代码的质量经常被忽视。这是一个项目的重要组成部分。您将希望能够移交您的实验，因此代码应该是可重用的。您还需要模型是可重现的，这不仅是出于学术原因，而且还需要出于商业目的重新构建模型。您可能还想在某个时候返回项目以改进模型。

   使建模项目可重用的一种常见方法是构建一个笔记本或一组笔记本。我们不会在本章中介绍它，因为建模项目很简单。

2. 构建脚本

   该脚本将采用一个参数，即 JSON 格式的评论路径——每行一个 JSON 格式的评论。它将输出一份关于评论分布的 JSON 格式的报告。

   以下是脚本的验收标准：

   • 它应该有一个有用的使用输出
   • 它应该在不到 1 分钟的时间内运行（适用于 95% 的电影）
   • 它应该以以下格式输出文件

   {
       "count": ###,
       "mean": 0.###,
       "std": 0.###,
       "median": 0.###,
       "min": 0.###,
       "max": 0.###,
   }

   我们将取平均值的分数应该是 0 到 1 之间的浮点数。许多分类器输出预测概率，但这对脚本的输出做了假设。

既然我们有了一个计划，就让我们执行它。

实施解决方案

回想一下我们在第 7 章中讨论的建模项目的步骤。让我们在这里看看它们。

1. 获取数据。
2. 看看数据。

3. 处理数据。

   我们已经有了数据，并且已经查看了它。让我们做一些基本的处理并存储它，以便我们可以更快地迭代我们的模型。

   首先，让我们将正面和负面组合成两个数据集，训练和测试。

   train = pos_train.unionAll(neg_train)
   test = pos_test.unionAll(neg_test)

   现在，让我们使用 Spark NLP 来处理数据。我们将保存词形化和规范化的标记，以及 GloVe 嵌入。这样，我们可以尝试不同的功能。

   让我们创建我们的管道。

   assembler = DocumentAssembler()\
       .setInputCol('text')\
       .setOutputCol('document')
   sentence = SentenceDetector() \
       .setInputCols(["document"]) \
       .setOutputCol("sentences")
   tokenizer = Tokenizer()\
       .setInputCols(['sentences'])\
       .setOutputCol('tokens')
   lemmatizer = LemmatizerModel.pretrained()\
       .setInputCols(['tokens'])\
       .setOutputCol('lemmas')
   normalizer = Normalizer()\
       .setCleanupPatterns([
           '[^a-zA-Z.-]+', 
           '^[^a-zA-Z]+', 
           '[^a-zA-Z]+$',
       ])\
       .setInputCols(['lemmas'])\
       .setOutputCol('normalized')\
       .setLowercase(True)
   glove = WordEmbeddingsModel.pretrained(name='glove_100d') \
       .setInputCols(['document', 'normalized']) \
       .setOutputCol('embeddings') \
   
   nlp_pipeline = Pipeline().setStages([
       assembler, sentence, tokenizer, 
       lemmatizer, normalizer, glove
   ]).fit(train)

   让我们只选择我们感兴趣的值——即原始数据加上规范化的标记和嵌入。

   train = nlp_pipeline.transform(train) \
       .selectExpr(
           'path', 'text', 'label', 
           'normalized.result AS normalized', 
           'embeddings.embeddings'
       )
   
   test = nlp_pipeline.transform(test) \
       .selectExpr(
           'path', 'text', 'label', 
           'normalized.result AS normalized', 
           'embeddings.embeddings'
       )
   nlp_pipel

   nlp_pipeline.write().overwrite().save('nlp_pipeline.3.12')

   回想一下我们在第 11 章中介绍的最简单的 doc2vec 版本，其中我们平均文档中的词向量以创建文档向量。我们将在这里使用这种技术。

   import numpy as np
   from pyspark.sql.types import *
   from pyspark.ml.linalg import DenseVector, VectorUDT
   
   def avg_wordvecs_fun(wordvecs):
       return DenseVector(np.mean(wordvecs, axis=0))
   
   avg_wordvecs = spark.udf.register(
       'avg_wordvecs', 
       avg_wordvecs_fun, 
       returnType=VectorUDT())
   
   train = train.withColumn('avg_wordvec', avg_wordvecs('embeddings'))
   test = test.withColumn('avg_wordvec', avg_wordvecs('embeddings'))
   train.drop('embeddings')
   test.drop('embeddings')

   现在，我们将它保存为镶木地板文件。这将让我们释放一些内存。

   train.write.mode('overwrite').parquet('imdb.train')
   test.write.mode('overwrite').parquet('imdb.test')

   让我们清理我们之前保存的数据，以便有更多的内存可以使用。

   pos_train.unpersist()
   neg_train.unpersist()
   pos_test.unpersist()
   neg_test.unpersist()

   现在我们加载我们的数据并持久化。

   train = spark.read.parquet('imdb.train').persist()
   test = spark.read.parquet('imdb.test').persist()

4. 特色化(Featurize)

   让我们看看我们的模型仅使用 TF.IDF 特征的效果如何。

   from pyspark.ml.feature import CountVectorizer, IDF

   tf = CountVectorizer()\
       .setInputCol('normalized')\
       .setOutputCol('tf')
   idf = IDF()\
       .setInputCol('tf')\
       .setOutputCol('tfidf')
   
   featurizer = Pipeline().setStages([tf, idf])

5. 模型(Model)

   现在我们有了我们的功能，我们可以构建我们的第一个模型。让我们从逻辑回归开始，这通常是一个很好的基线。

   from pyspark.ml.feature import VectorAssembler
   from pyspark.ml.classification import LogisticRegression

   vec_assembler = VectorAssembler()\
       .setInputCols(['avg_wordvec'])\
       .setOutputCol('features')
   logreg = LogisticRegression()\
       .setFeaturesCol('features')\
       .setLabelCol('label')
   
   model_pipeline = Pipeline()\
       .setStages([featurizer, vec_assembler, logreg])

   model = model_pipeline.fit(train)

   现在让我们保存模型。

   model.write().overwrite().save('model.3.12')

   现在我们已经拟合了一个模型，让我们得到我们的预测。

   train_preds = model.transform(train)

   test_preds = model.transform(test)

6. 评估

   让我们计算一下我们在训练和测试中的 F1 分数。

   from pyspark.ml.evaluation import MulticlassClassificationEvaluator

   evaluator = MulticlassClassificationEvaluator()\
       .setMetricName('f1')

   evaluator.evaluate(train_preds)

   0.8029598474121058

   evaluator.evaluate(test_preds)

   0.8010723532212578

   这超出了最低验收标准，因此我们已准备好发布此模型。

7. 审查

   当然，我们可以找出改进模型的方法。但重要的是要推出第一个版本。在我们部署了初始版本之后，我们可以开始寻找改进模型的方法。

8. 部署

   对于此应用程序，部署只是使脚本可用。实际上，离线“部署”通常涉及创建可以按需或定期运行的工作流。对于此应用程序，只需将脚本放在可以运行以进行新评论的地方即可。

   %%writefile movie_review_analysis.py
   
   """
   此脚本获取包含相同评论的文件。
   它将分析结果输出到 std.out。
   """
   
   
   import argparse as ap
   import json
   from pyspark.sql import SparkSession
   from pyspark.ml import PipelineModel
   
   
   if __name__ == '__main__':
       print('beginning...')
       parser = ap.ArgumentParser(description='Movie Review Analysis')
       parser.add_argument('-file', metavar='DATA', type=str,
                           required=True, 
                           help='The file containing the reviews '\
                                'in JSON format, one JSON review '\
                                'per line')
       
       options = vars(parser.parse_args())
       
       spark = SparkSession.builder \
           .master("local[*]") \
           .appName("Movie Analysis") \
           .config("spark.driver.memory", "12g") \
           .config("spark.executor.memory", "12g") \
           .config("spark.jars.packages", 
                   "JohnSnowLabs:spark-nlp:2.2.2") \
           .getOrCreate()
       
       nlp_pipeline = PipelineModel.load('nlp_pipeline.3.12')
       model = PipelineModel.load('model.3.12')
       
       data = spark.read.json(options['file'])
       
       nlp_procd = nlp_pipeline.transform(data)
       preds = model.transform(nlp_procd)
       
       results = preds.selectExpr(
           'count(*)',
           'mean(rawPrediction[1])',
           'std(rawPrediction[1])',
           'median(rawPrediction[1])',
           'min(rawPrediction[1])',
           'max(rawPrediction[1])',
       ).first().asDict()
       
       print(json.dump(results))
   

   此脚本可用于获取一组评论并将其汇总为单个分数以及一些附加统计信息。

测试和测量解决方案

现在我们有了应用程序的第一个实现，让我们来谈谈指标。在更现实的情况下，您将在计划阶段定义您的指标。但是，一旦我们有了具体的参考资料，就更容易解释其中的一些主题。

审查

审查过程是编写任何应用程序的另一个重要部分。开发人员很容易在自己的项目中出现盲点。这就是为什么我们必须请其他人来审查我们的工作。由于技术和人为原因，此过程可能很困难。任何审查过程中最重要的部分是它不是个人的。审阅者和开发人员都应该以协作为目标来处理流程。如果发现问题，那就是机会。开发人员避免了以后的问题，审阅者可以学到一些可以帮助他们避免未来工作中的问题的东西。

说说复习的步骤吧。

1. 架构审查：这是其他工程师、产品所有者和利益相关者审查应用程序将如何部署的地方。这应该在发展规划阶段结束时完成
2. 模型审查：当开发人员或数据科学家拥有一个他们认为能够满足利益相关者期望的模型时，就会这样做。该模型应该与其他数据科学家或熟悉机器学习概念的人一起审查，并且应该与利益相关者一起进行另一次审查。技术评审应涵盖项目的数据、处理、建模和测量方面。非技术审查应解释模型的假设和限制，以验证它是否符合利益相关者的期望。
3. 代码审查：这对于任何软件应用程序都是必要的。代码应该由对项目有一定了解的人审查。如果审阅者没有应用程序的上下文，他们将很难或不可能发现代码中的逻辑错误。

在我们的情况下，这个应用程序非常简单。我们只是在开发一个脚本，所以没有实际的架构可供审查。但是，将其部署为脚本的计划必须由利益相关者审查。模型审查也很简单，因为我们有一个干净的数据集和简单的模型。一般不会出现这种情况。同样，我们的脚本也很简单。代码审查可能会建议我们开发一个小型测试数据集，以确保模型能够运行我们期望的数据。

这些审查发生在模型的开发过程中。一旦我们准备好部署，我们应该进行更多的审查，以确保我们已经为部署做好了准备。

结论

现在我们已经构建了我们的第一个应用程序。这是一个简单的应用程序，但它让我们了解了很多关于如何部署更复杂的应用程序的知识。在下一章中，我们将再次查看离线应用程序，但这不仅仅基于模型的输出。我们将构建一个可以查询的本体。

许多基于 Spark 的应用程序都是这样的离线工具。如果我们想为服务背后的实时模型提供服务，我们需要寻找其他地方。有几种选择，我们将在第 19 章讨论。

情绪分析是一项引人入胜的任务，它使用我们已经介绍过的工具和技术。还有更复杂的示例，但通过使用良好的开发流程，我们总能开发出一个简单的应用程序。