如何使用Apache Spark中的实时机器学习管道构建预测模型?

    1-main-image.jpg


    在前一篇题为“使用Apache Kafka和Spark的实时数据管道”的文章中，我描述了如何构建高吞吐量，可扩展，可靠且容错的数据管道，能够获取基于事件的数据并最终将这些事件流式传输到我们处理它们的Apache Spark。我通过简单地使用Apache Spark来使用基于事件的数据并将它们打印到控制台来结束上一篇文章。在我上一篇题为“带有Avro Bijection的Apache Kafka Producer”的文章中，我更新了我们的数据管道，使用Apache Kafka Producer在Twitter4j包装器中使用Twitter的Hosebird客户端收集推文，其中推文使用Avro库和Bijection进行序列化。在本文中，我们将使用更新的基于事件的数据管道，使用Apache Spark实时构建预测模型。
    目标
    我们在本文中的目标是对有关航空公司的推文进行实时情绪分析 - 特别是考虑到英国航空公司最近的全球IT故障！我们将尝试自动将关于航空公司的推文分类为对该航空公司的负面或非负面因为我们的数据管道几乎实时接收它们。
    数据
    我们将用于构建我们的训练模型的数据集是来自CrowdFlower的 2015年2月美国航空公司的14,640条推文的集合。这些推文已经被分类，以确定他们的文字内容是否对所涉及的航空公司是积极的，消极的或中立的。可以使用以下链接下载数据集。
    基础
    让我们首先解释一些建模概念背后的基本理论，我将在本文后面介绍。我们知识库背后的首要目标是让尽可能多的人能够获得有关大数据和数据分析的知识。话虽如此，虽然本节旨在介绍一些更基本的概念，但很难在一页中详细解释数学理论。因此，如果这篇文章引起您的兴趣（我希望它会！），我建议您查看一些在Coursera和edX等网站上提供的精彩在线课程，或者甚至注册更正式的课程。像开放大学这样的组织。
    预测模型
    建立预测模型以进行预测！监督学习是一种机器学习算法，它使用训练数据集根据已知结果训练模型。然后将该模型应用于之前未见过的新数据进行预测。这些预测可以采用分类预测的形式（例如，是/否，正/负/中性等） - 称为分类算法 - 或连续预测（例如价格，权重等） - 称为回归算法。示例监督学习算法包括线性回归，逻辑回归以及分类和回归树。在本文中，我将介绍后者 - 分类和回归树 - 并展示我们如何使用这些算法来训练模型，以根据我们上面下载的训练数据集对我们的推文进行分类。
    分类和回归树（CART）
    分类和回归树（CART）模型生成可以遍历以预测结果的树。对于那些熟悉线性回归和Logistic回归的人来说，CART不假设像线性回归这样的线性模型，与Logistic回归相比，它更容易理解和解释。以下来自维基百科的图片显示了一个简单的树，它可以帮助我们根据一些属性或属性来预测您是否可能作为泰坦尼克号的乘客生存，其中每个叶子下方的数字显示生存概率和百分比该叶子中的观察结果。 cart-example.png
    CART模型旨在将数据拆分为同类子集 - 您可以从CART模型中得出的基本预测因此只占每个子集中的大多数。根据您现有的CART模型，新的观察结果将属于一个子集，您可以只预测该子集中的现有多数。树有助于使CART模型具有可解释性 - 您可以从树的顶部开始向下移动，向左移动以获得积极响应，向右移动以获得否定响应。顾名思义，CART模型允许您预测连续结果或分类结果。在本文中，我们将使用CART模型来帮助我们将推文归类为对航空公司的否定或非负面即分类问题。
    随机森林
    随机森林旨在通过构建多个CART树（即森林）来进一步提高CART模型的准确性。使用随机森林的缺点是您的模型变得不易理解且解释性较差，但它们可能会提高预测的准确性。为了预测新观察，森林中的每棵树都投票在结果上你可以简单地选择获得最多选票的结果。为了构建随机森林，每个组成树被迫仅从一个自举的数据样本中拆分可用自变量的随机子集。例如，每个树的训练数据随机选择替换 - 因为每个树使用不同的自变量和不同的训练数据，我们生成不同树的森林。下图概述了随机森林背后的基本原理。 random-forest-overview.jpg 分类矩阵
    分类（或混淆）矩阵可以帮助我们量化预测模型的准确性。通过将模型（列）中的预测值与实际值（行）进行比较来构建分类矩阵，如下所示： confusion-matrix.png 然后，Matrix中的每个单元格都会提供属于该类别的观察数量。TRUE Negatives和TRUE Positives是模型正确预测的观测数。FALSE Negatives和FALSE Positives是模型预测错误的观察数量。因此，这产生了许多可量化的度量，使我们能够说明模型与其他模型相比的准确度： - 灵敏度或真阳性率 = TP /（TP + FN）测量模型正确预测的阳性结果的百分比。 - 特异性或真阴性率 = TN /（TN + FP）测量模型正确预测的阴性结果的百分比。 - 总体准确度 =（TN + TP）/ N测量给定N个观测值的模型的总体准确度。 - 总误差率 =（FP + FN）/ N测量给定N个观测值的模型的总误差率。 自然语言处理（NLP）
    推文显然是文本性质的。这意味着它们不符合任何预定义的结构，并且可能包括拼写错误，俚语和其他形式的非传统语法。此外，每天都会产生数以亿计的推文，这使得人们无法进行大量的分析。自然语言处理（NLP）是使用计算机理解文本的领域 - 理解人类语言并从中获得意义。在我们将要构建的模型中，单词本身将是我们在CART模型中使用的变量，用于对我们的推文背后的情绪进行分类！
    常用术语
    以下是我的解释中经常出现的一些常用术语供您参考： - 文档 - 文档是单词的定义容器。文档可以是单个记录，文章或推文。 - 语料库 - 语料库是文档的集合。 - 术语频率 - 术语在文档中出现的次数。 - 反向文档频率 - 衡量一个术语在语料库中是常见的还是罕见的度量。例如，术语的可以在给定的语料库出现无数次。仅使用术语频率，该术语可以被赋予与其频率成比例的权重，而其他更有意义的术语不一定被赋予更多权重。逆文档频率通过考虑更广泛的语料库来抵消术语的简单频率。 预处理数据
    NLP的基本特征之一是在构建预测模型之前预处理文本数据的想法，以尽可能地标准化来自不同源的文本。常见的预处理技术包括： - Common Casing - 将所有文本转换为小写或大写。 - 无意义的标点符号 - 基本方法是删除所有非字母数字字符。但要小心，因为一些非字母数字字符是有意义的，例如推文中的@和＃。因此，应根据您尝试回答的问题来定义删除的非字母数字字符集。 - 停止词语 - 诸如I，this，there等词语用于表达句子以表达其含义，但不一定有助于确定潜在的情绪。这些停用词通常从文本中删除，还有减少数据集大小的额外好处。 - 词干 - 这是将常用词减少到共同词干的过程。例如，词语连接，连接，结缔组织，连接和连接都可以被减少到其的共用杆连接而不去除潜在的意思。这并不是说Stemming没有错误 - 事实上，词干算法容易出错。然而，为了建立预测模型并进一步减小数据集的大小，这是一种有价值的技术。 - Lemmatisation - 虽然Stemming迅速将单词缩减为基本形式，但它没有考虑上下文，因此无法区分具有不同含义的单词，这取决于它在句子或上下文中的位置。Lemmatisation并没有粗略地简化基于共同词干的单词，而是旨在删除屈折结尾只是为了返回一个单词的词典形式，称为引理。例如，词语是，被，被和被可以减小到引理是虽然Stemmer无法推断出这种语境意义。虽然Lemmatisation可用于更好地保留背景和意义，但它会带来额外的复杂性和时间。 一袋字
    在我们预处理了数据之后，我们如何从仅仅文本中识别出用于我们的预测模型的自变量？好吧，一种方法叫做Bag of Words。在这种方法中，您只需计算每个单词在文本中出现的次数（即该单词的频率），从而为每个单词生成一个可用作基线的功能。然后，每个字频配对成为我们可以在模型中使用的独立变量。
    术语频率 - 逆文档频率（TF-IDF）
    术语频率 - 反向文档频率（TF-IDF）不仅仅是简单地计算每个单词的频率，而是旨在提供一个单词考虑单词在整个数据集中出现的频率的重要性。TF-IDF度量与单词在文档中出现的次数成比例增加，与整个数据集中单词的频率相抵消。
    特征向量
    通过将文本转换为一系列词频配对（如Bag of Words方法）或一系列单词重要性配对（与TF-IDF方法一样），我们实际上将文本转换为称为特征的数字向量向量。这很重要，因为正是这些特征向量被传递到机器学习算法以训练和测试监督预测模型。
    Apache Spark机器学习管道
    在我们开始构建预测模型之前，我想解释一下Apache Spark的机器学习库中的一些核心组件，它们允许我们构建预测模型。 - Transformer - Spark中的Transformer实际上将DataFrame作为其输入并生成新的DataFrame，通常是通过将新列附加到原始DataFrame作为读取和映射操作的结果。例如，在学习模型中，输入DataFrame可以包含包含特征向量的列。然后，变换器可以读取该列并预测每个特征向量的结果（标签），生成包含预测标签的新列。 - Estimator - Spark中的Estimator本质上是一种学习算法，可以训练数据生成模型 - 模型是变换器。例如，可以调用Spark Estimator LogisticRegression来训练训练数据集，生成LogisticRegressionModel，它是生成的Model和Transformer。 - 管道 - Spark中的管道是一系列有序的阶段，每个阶段都是变压器或估算器。 典型的Training Pipeline如下所示： spark-ml-pipeline.png 在上面的示例中，原始文本被拆分以生成每个文档的单词序列。使用某种术语频率算法从这些序列创建特征向量，然后将其传递给估计器以训练模型。
    生成的测试管道可能如下所示： spark-ml-pipeline-model.png 这里，再次拆分新的未见原始文本以生成每个文档的单词序列，并遵循相同的处理步骤。但这一次，上面生成的估计模型应用于新的特征向量以生成预测。通过使用这些管道和管道模型，我们可以构建训练模型，然后将相同的过程应用于我们的基于实时事件的数据流！
    Apache Maven
    现在我们已经介绍了一些基本概念，让我们开始编码！与前面的文章一样，我们将使用Apache Maven来处理构建依赖项。我的POM中需要注意的几件事情是： - ApacheSpark提供的依赖 -我已经被标记为Apache的Spark依赖性提供，因为它们包含在Apache的Spark实例。 - Stanford NLP - 我将使用Stanford NLP Library对文本进行词形还原。这个库是用Java编写的，因此可以在Java和Scala Spark应用程序中使用（我将在Scala中编写我们的机器学习管道）。 ```xml 4.0.0 
    1. <groupId>io.keisan.knowledgebase.spark.mllib</groupId>
    2. <artifactId>keisan-spark-mllib</artifactId>
    3. <version>0.0.1-SNAPSHOT</version>
    4. <packaging>jar</packaging>
    6. <!-- Project Information -->
    7. <name>Example Spark Machine Learning Pipelines</name>
    8. <description>Keisan Knowledgebase Spark MLlib Project - Example Spark Machine Learning Pipelines</description>
    9. <url>https://www.keisan.io/knowledgebase/real-time-ml-pipeline-spark</url>
    10. <organization>
    11.     <name>Keisan Ltd</name>
    12.     <url>https://www.keisan.io</url>
    13. </organization>
    14. <developers>
    15.     <developer>
    16.         <id>jillur.quddus</id>
    17.         <name>Jillur Quddus</name>
    18.         <email>contactus@keisan.io</email>
    19.         <url>https://www.keisan.io</url>
    20.         <organization>Keisan Ltd</organization>
    21.         <organizationUrl>https://www.keisan.io</organizationUrl>
    22.         <roles>
    23.             <role>Lead Engineer</role>
    24.             <role>Data Scientist</role>
    25.         </roles>
    26.         <timezone>Europe/London</timezone>
    27.     </developer>
    28. </developers>
    30. <!-- Properties -->
    31. <properties>
    32.     <apache.spark.core.2.11.version>2.1.0</apache.spark.core.2.11.version>
    33.     <apache.spark.mllib.2.11.version>2.1.0</apache.spark.mllib.2.11.version>
    34.     <apache.spark.sql.2.11.version>2.1.0</apache.spark.sql.2.11.version>
    35.     <jdk.version>1.8</jdk.version>
    36.     <maven.plugins.maven-assembly-plugin.version>3.0.0</maven.plugins.maven-assembly-plugin.version>
    37.     <maven.plugins.maven-compiler-plugin.version>3.6.1</maven.plugins.maven-compiler-plugin.version>
    38.     <output.directory>/keisan/knowledgebase/spark/mllib/jars</output.directory>
    39.     <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    40.     <scala.library.version>2.11.8</scala.library.version>
    41.     <scala.maven.plugin.version>3.2.2</scala.maven.plugin.version>
    42.     <stanford.corenlp.version>3.7.0</stanford.corenlp.version>
    43. </properties>
    45. <!-- Dependencies -->
    46. <dependencies>
    48.     <!-- Apache Spark -->
    49.     <dependency>
    50.         <groupId>org.apache.spark</groupId>
    51.         <artifactId>spark-core_2.11</artifactId>
    52.         <version>${apache.spark.core.2.11.version}</version>
    53.         <scope>provided</scope>
    54.     </dependency>
    55.     <dependency>
    56.         <groupId>org.apache.spark</groupId>
    57.         <artifactId>spark-sql_2.11</artifactId>
    58.         <version>${apache.spark.sql.2.11.version}</version>
    59.         <scope>provided</scope>
    60.     </dependency>
    61.     <dependency>
    62.         <groupId>org.apache.spark</groupId>
    63.         <artifactId>spark-mllib_2.11</artifactId>
    64.         <version>${apache.spark.mllib.2.11.version}</version>
    65.         <scope>provided</scope>
    66.     </dependency>
    68.     <!-- Scala -->
    69.     <dependency>
    70.         <groupId>org.scala-lang</groupId>
    71.         <artifactId>scala-library</artifactId>
    72.         <version>${scala.library.version}</version>
    73.     </dependency>
    75.     <!-- Stanford NLP -->
    76.     <dependency>
    77.         <groupId>edu.stanford.nlp</groupId>
    78.         <artifactId>stanford-corenlp</artifactId>
    79.         <version>${stanford.corenlp.version}</version>
    80.     </dependency>
    81.     <dependency>
    82.         <groupId>edu.stanford.nlp</groupId>
    83.         <artifactId>stanford-corenlp</artifactId>
    84.         <version>${stanford.corenlp.version}</version>
    85.         <classifier>models</classifier>
    86.     </dependency>
    88. </dependencies>
    90. <!-- Project Builder -->
    91. <build>
    93.     <!-- Plugins -->
    94.     <plugins>
    96.         <!-- Maven Compiler: Compile the Sources of the Project -->
    97.         <plugin>
    98.             <groupId>org.apache.maven.plugins</groupId>
    99.             <artifactId>maven-compiler-plugin</artifactId>
    100.             <version>${maven.plugins.maven-compiler-plugin.version}</version>
    101.             <configuration>
    102.                 <source>${jdk.version}</source>
    103.                 <target>${jdk.version}</target>
    104.             </configuration>
    105.         </plugin>
    107.         <!-- Maven Assembly: Aggregate project output with its dependencies -->
    108.         <plugin>
    109.             <groupId>org.apache.maven.plugins</groupId>
    110.             <artifactId>maven-assembly-plugin</artifactId>
    111.             <version>${maven.plugins.maven-assembly-plugin.version}</version>
    112.             <configuration>
    114.                 <!-- Final JAR Filename -->
    115.                 <finalName>keisan-spark-mllib-${project.version}</finalName>
    116.                 <appendAssemblyId>false</appendAssemblyId>
    118.                 <!-- Include all Project Dependencies -->
    119.                 <descriptorRefs>
    120.                     <descriptorRef>jar-with-dependencies</descriptorRef>
    121.                 </descriptorRefs>
    123.                 <!-- JAR with dependencies Output Target Directory -->
    124.                 <outputDirectory>${output.directory}</outputDirectory>
    126.             </configuration>
    127.             <executions>
    129.                 <!-- Bind the assembly to the package phase -->
    130.                 <execution>
    131.                     <id>make-assembly</id>
    132.                     <phase>package</phase>
    133.                     <goals>
    134.                         <goal>single</goal>
    135.                     </goals>
    136.                 </execution>
    138.             </executions>
    140.         </plugin>
    142.         <!-- Scala Maven Plugin -->
    143.         <plugin>
    144.             <groupId>net.alchim31.maven</groupId>
    145.             <artifactId>scala-maven-plugin</artifactId>
    146.             <version>${scala.maven.plugin.version}</version>
    147.         </plugin>
    149.     </plugins>
    151. </build>

    2. **预处理管道**<br />正如我上面提到的，我将使用[Scala](https://www.scala-lang.org/)编写我的Spark Machine学习管道。Scala在JVM上运行，因此允许我们在Scala应用程序中使用用Java编写的库。此外，与Java相比，Scala在开发Spark应用程序时往往不那么冗长，而且对于新数据科学家来说更容易学习。<br />我要执行的第一个编码是在Scala中定义一系列函数，这些函数将允许我们预处理Spark Dataframes中的文本。如上所述，我将定义一系列函数，这些函数将允许我们小写文本，删除无意义的标点符号，删除停用词和lemmatize文本（使用Stanford Core NLP库）。
    4. ```scala
    5. package io.keisan.knowledgebase.spark.mllib;
    7. import java.util.Properties;
    9. import edu.stanford.nlp.ling.CoreAnnotations.LemmaAnnotation;
    10. import edu.stanford.nlp.ling.CoreAnnotations.SentencesAnnotation;
    11. import edu.stanford.nlp.ling.CoreAnnotations.TokensAnnotation;
    12. import edu.stanford.nlp.pipeline.Annotation;
    13. import edu.stanford.nlp.pipeline.StanfordCoreNLP;
    14. import edu.stanford.nlp.util.CoreMap;
    16. import org.apache.spark.sql.Dataset;
    17. import org.apache.spark.sql.Row;
    18. import org.apache.spark.sql.functions._;
    19. import org.apache.spark.sql.types._;
    21. import org.apache.spark.ml.feature.StopWordsRemover;
    23. import scala.collection.JavaConversions._;
    24. import scala.collection.mutable.ArrayBuffer;
    26. /**
    27.  * Pre-Processor Utilities / Helper Functions
    28.  * A collection of functions to pre-process text
    29.  *
    30.  * @author jillur.quddus
    31.  * @version 0.0.1
    32.  */
    34. object PreProcessorUtils {
    36.   /**
    37.    * Lowercase and Remove Punctuation
    38.    * Lowercase and remove non-alphanumeric-space characters from the text field
    39.    *
    40.    * @param corpus The collection of documents as a Dataframe
    41.    * @param textColumn The name of the column containing the text to be pre-processed
    42.    * @return A Dataframe with the text lowercased and non-alphanumeric-space characters removed
    43.    *
    44.    */
    46.   def lowercaseRemovePunctuation(corpus:Dataset[Row], textColumn:String): Dataset[Row] = {
    47.     return corpus.withColumn(textColumn, regexp_replace(lower(corpus(textColumn)), "[^a-zA-Z0-9 ]", ""));
    48.   }
    50.   /**
    51.    * Text Lemmatizer
    52.    * Given a text string, generate a sequence of Lemmas
    53.    *
    54.    * @param text The text string to lemmatize
    55.    * @param pipeline The Stanford Core NLP Pipeline
    56.    * @return A sequence of lemmas
    57.    *
    58.    */
    60.   def lemmatizeText(text: String, pipeline:StanfordCoreNLP): Seq[String] = {
    62.     val doc = new Annotation(text);
    63.     pipeline.annotate(doc);
    64.     val lemmas = new ArrayBuffer[String]();
    65.     val sentences = doc.get(classOf[SentencesAnnotation]);
    66.     for (sentence <- sentences; token <- sentence.get(classOf[TokensAnnotation])) {
    67.       val lemma = token.get(classOf[LemmaAnnotation])
    68.       if (lemma.length > 2 && isOnlyLetters(lemma)) {
    69.         lemmas += lemma.toLowerCase
    70.       }
    71.     }
    73.     return lemmas;
    75.   }
    77.   /**
    78.    * Check that a given String is made up entirely of alpha characters
    79.    *
    80.    * @param str The string to test for alpha characters
    81.    * @return Boolean
    82.    */
    84.   def isOnlyLetters(str: String) = str.forall(c => Character.isLetter(c));
    86.   /**
    87.    * Stop Words Remover
    88.    * Remove Stop Words from a given input column containing a sequence of String
    89.    *
    90.    * @param corpus The collection of documents as a Dataframe
    91.    * @param inputColumn The name of the column containing a sequence of Strings to filter
    92.    * @param outputColumn The name of the column to output the filtered sequence of Strings to
    93.    */
    95.   def stopWordRemover(corpus:Dataset[Row], inputColumn:String, outputColumn:String): Dataset[Row] = {
    97.     val stopWordsRemover = new StopWordsRemover()
    98.                              .setInputCol(inputColumn)
    99.                              .setOutputCol(outputColumn);
    100.     return stopWordsRemover.transform(corpus);
    102.   }
    104. }

    摄取，标记和预处理数据
    我们现在可以提取我们从CrowdFlower下载的数据。我们已经下载的数据包含了标签栏（即）的结果，即airline_sentiment要么是正面，负面或中立的。我们可以保留标签，即多分类。但是，我们通过使用二进制分类标记数据（负面或非负面）来使其更简单一些。然后，我们的模型将能够预测推文是否只是对有问题的航空公司是负面的。要做到这一点，在我的代码下面我创建了一个新的String列，被称为negative_sentiment_label要么是真正的负情绪推特（我们取得积极成果）或在所有其他情况下都是假的（我们的负面结果，其中推文被分类为情绪的自然或积极）。在我们标记了数据之后，我们可以应用上面定义的预处理技术来删除非含义标点符号，将我们的文本解释并删除停用词。

    1. object TrainDecisionTreeClassifier {
    3.   def main(args: Array[String]) = {
    5.     /********************************************************************
    6.      * SPARK CONTEXT
    7.      ********************************************************************/
    9.     // Create the Spark Context
    10.     val conf = new SparkConf()
    11.       .setAppName("Sentiment Models")
    12.             .setMaster("spark://<Spark Master Hostname>:7077");
    13.     val sc = new SparkContext(conf);
    14.     val sparkSession = SparkSession.builder().getOrCreate();
    15.     import sparkSession.implicits._;
    17.     /********************************************************************
    18.      * INGEST THE CORPUS
    19.      ********************************************************************/
    21.     // Define the CSV Dataset Schema
    22.     val schema = new StructType(Array(
    23.                     StructField("unit_id", LongType, true),
    24.                     StructField("golden", BooleanType, true),
    25.                     StructField("unit_state", StringType, true),
    26.                     StructField("trusted_judgements", IntegerType, true),
    27.                     StructField("last_judgement_at", StringType, true),
    28.                     StructField("airline_sentiment", StringType, true),
    29.                     StructField("airline_sentiment_confidence", DoubleType, true),
    30.                     StructField("negative_reason", StringType, true),
    31.                     StructField("negative_reason_confidence", DoubleType, true),
    32.                     StructField("airline", StringType, true),
    33.                     StructField("airline_sentiment_gold", StringType, true),
    34.                     StructField("name", StringType, true),
    35.                     StructField("negative_reason_gold", StringType, true),
    36.                     StructField("retweet_count", IntegerType, true),
    37.                     StructField("text", StringType, true),
    38.                     StructField("tweet_coordinates", StringType, true),
    39.                     StructField("tweet_created", StringType, true),
    40.                     StructField("tweet_id", StringType, true),
    41.                     StructField("tweet_location", StringType, true),
    42.                     StructField("user_timezone", StringType, true)
    43.                  ));
    45.     // Read the CSV Dataset, keeping only those columns that we need to build our model
    46.     var tweetsDF = SparkSession.builder().getOrCreate().read
    47.                       .format("csv")
    48.                       .option("header", true)
    49.                       .option("delimiter", ",")
    50.                       .option("mode", "DROPMALFORMED")
    51.                       .schema(schema)
    52.                       .load("hdfs://<Namenode Hostname>:9000/keisan/knowledgebase/spark/mllib/raw/Airline-Sentiment-2-w-AA.csv")
    53.                       .select("airline_sentiment", "text");
    55.     /********************************************************************
    56.      * LABEL THE DATA
    57.      ********************************************************************/
    59.     // We are interested in detecting tweets with negative sentiment. Let us create a new column whereby
    60.     // if the sentiment is negative, this new column is TRUE (Positive Outcome), and FALSE (Negative Outcome)
    61.     // in all other cases
    63.     tweetsDF = tweetsDF.withColumn("negative_sentiment_label", when(tweetsDF("airline_sentiment") === "negative", lit("true")).otherwise(lit("false")))
    64.                 .select("text", "negative_sentiment_label");
    66.     /********************************************************************
    67.      * APPLY THE PRE-PROCESSING PIPELINE
    68.      ********************************************************************/
    70.     // Let us now perform some simple pre-processing including converting the text column to lowercase
    71.     // and removing all non-alphanumeric characters
    73.     val lowercasedDF = PreProcessorUtils.lowercaseRemovePunctuation(tweetsDF, "text");
    75.     // Lemmatize the text to generate a sequence of Lemmas using the Stanford NLP Library
    76.     // By using mapPartitions, we create the Stanford NLP Pipeline once per partition rather than once per RDD entry
    78.     val lemmatizedDF = lowercasedDF.select("text", "negative_sentiment_label").rdd.mapPartitions(p => {
    80.       // Define the NLP Lemmatizer Pipeline once per partition
    81.       val props = new Properties();
    82.       props.put("annotators", "tokenize, ssplit, pos, lemma");
    83.       val pipeline = new StanfordCoreNLP(props);
    85.       // Lemmatize the text and preserve the Negative Sentiment Label
    86.       p.map{
    87.         case Row(text: String, negative_sentiment_label:String) => (PreProcessorUtils.lemmatizeText(text, pipeline), negative_sentiment_label);
    88.       };
    90.     }).toDF("lemmas", "negative_sentiment_label");
    92.     // Remove Stop Words from the sequence of Lemmas
    93.     val stopWordsRemovedDF = PreProcessorUtils.stopWordRemover(lemmatizedDF, "lemmas", "filtered_lemmas")
    94.                                 .where(size(col("filtered_lemmas")) > 1);

    现在，如果你运行动作stopWordsRemovedDF.show（false），你应该看到一个类似于下面的DataFrame：

    preprocessed-training-data.png

    希望您应该看到3列 - 从原始文本派生的词条序列，删除了停用词的过滤的词条序列，以及我们的标签列，对于负面情绪推文是真的，对于所有其他情况都是假的。
    特征向量和训练模型
    我们现在准备生成我们的训练模型。第一步是将我们过滤的引理序列转换为特征向量，以便我们可以将它们传递给Estimator以生成我们的训练模型。如上所述，我们将使用TF-IDF方法生成特征向量，其中考虑术语对语料库中文档的重要性，而不仅仅考虑术语频率（TF）。
    为了使事情更正式，我们可以说术语频率TF（t，d）是术语t出现在文档d中的次数。我们可以说文档频率DF（t，D）是我们的语料库D中包含术语t的文档数。然而，正如我上面所描述的那样，简单地使用TF可能过分强调经常出现但没有表达意义的术语。我们可以说反向文档频率IDF（t，D）是一个数字度量，通过考虑术语在语料库中出现的频率，术语的重要程度，可以计算如下：

    idf.png

    在这个等式中，| D | 是语料库D中的文档总数。并且由于使用了对数，如果一个术语出现在所有文档中，其IDF值将变为0.要计算最终的TF-IDF度量，您可以将TF乘以IDF，如下所示：

    tfidf.png

    在Spark的MLlib库中，Term Frequency TF向量可以使用Transformer HashingTF生成，它采用一系列术语并将它们转换为固定长度的特征向量。然后可以将这些特征向量传递给IDFModel，IDFModel由适合数据集的IDF估计器生成，以根据跨语料库的频率对列进行缩放。
    一旦我们从引理序列生成了缩放的特征向量，我们就可以将它们传递给Estimators来构建我们的训练模型。在本文中，我们将基于决策树分类器构建训练模型估计。在下面的代码中，我定义了一个新的Scala对象来保存模型助手函数。第一个函数将从一系列（过滤的）引理中生成缩放的特征向量。第二个函数将使用这些缩放的特征向量和先前定义的标签来训练和输出决策树分类器。第三个函数将再次使用这些缩放的特征向量和先前定义的标签来训练和输出随机森林分类器。第四个功能将允许我们生成指标，例如我们的模型精度，从应用我们的训练模型到测试数据。

    1. package io.keisan.knowledgebase.spark.mllib;
    3. import org.apache.spark.sql.Dataset;
    4. import org.apache.spark.sql.functions._;
    5. import org.apache.spark.sql.types._;
    6. import org.apache.spark.sql.Row;
    8. import org.apache.spark.ml.Pipeline;
    9. import org.apache.spark.ml.PipelineModel;
    10. import org.apache.spark.ml.classification.DecisionTreeClassifier;
    11. import org.apache.spark.ml.classification.RandomForestClassifier;
    12. import org.apache.spark.ml.feature.IndexToString;
    13. import org.apache.spark.ml.feature.StringIndexer;
    14. import org.apache.spark.ml.feature.HashingTF;
    15. import org.apache.spark.ml.feature.IDF;
    16. import org.apache.spark.mllib.evaluation.MulticlassMetrics;
    18. import scala.collection.JavaConversions._;
    20. /**
    21.  * Predictive Model Utilities / Helper Functions
    22.  * A collection of functions to build Predictive Models
    23.  *
    24.  * @author jillur.quddus
    25.  * @version 0.0.1
    26.  */
    28. object ModelUtils {
    30.   // Size of the fixed-length Feature Vectors
    31.   val numFeatures = 4096;
    33.   // Number of trees in our random forests
    34.   val numTrees = 256;
    36.   /**
    37.    * Term Frequency-Inverse Document Frequency (TF-IDF)
    38.    * Generate Term Frequency Feature Vectors by passing the sequence of lemmas to the HashingTF Transformer.
    39.    * Fit the IDF Estimator to the Featurized Dataset to generate the IDFModel.
    40.    * Pass the TF Feature Vectors to the IDFModel to scale based on frequency across the corpus
    41.    *
    42.    * @param corpus Dataset containing the sequence of lemmas
    43.    * @param inputColumn The name of the column containing the sequence of (filtered) lemmas
    44.    * @param outputColumn The name of the column to store the Scaled Feature Vectors
    45.    * @return A DataFrame with the Scaled Feature Vectors
    46.    *
    47.    */
    49.   def tfidf(corpus:Dataset[Row], inputColumn:String, outputColumn:String): Dataset[Row] = {
    51.     // Convert the sequence of Lemmas into fixed-length feature vectors using the HashingTF Transformer
    52.     val hashingTF = new HashingTF()
    53.                       .setInputCol(inputColumn)
    54.                       .setOutputCol("raw_features")
    55.                       .setNumFeatures(numFeatures);
    56.     val featurizedData = hashingTF.transform(corpus);
    58.     // Takes the feature vectors and scale each column based on how frequently it appears in the corpus
    59.     val idf = new IDF().setInputCol("raw_features").setOutputCol(outputColumn);
    60.     val idfModel = idf.fit(featurizedData);
    61.     return idfModel.transform(featurizedData);
    63.   }
    65.   /**
    66.    * Build a Decision Tree Classifier
    67.    * Train a Decision Tree Model by supplying the training dataset that includes the label and feature vector columns
    68.    *
    69.    * @param featuresDF The full DataFrame containing the labels and feature vectors
    70.    * @param trainingDF The training split DataFrame to be used to train the Model
    71.    * @param labelColumn The name of the column containing the labels
    72.    * @param featuresColumn The name of the column containing the scaled feature vectors
    73.    * @return The PipelineModel containing our trained decision tree model
    74.    *
    75.    */
    77.   def trainDecisionTreeModel(featurizedDF:Dataset[Row], trainingDF:Dataset[Row], labelColumn:String,
    78.       featuresColumn:String): PipelineModel = {
    80.     // Index the Labels
    81.     val labelIndexer = new StringIndexer()
    82.                           .setInputCol(labelColumn)
    83.                           .setOutputCol("indexed_label")
    84.                           .fit(featurizedDF);
    86.     // Define the Decision Tree Model
    87.     val decisionTreeModel = new DecisionTreeClassifier()
    88.                               .setLabelCol("indexed_label")
    89.                               .setFeaturesCol(featuresColumn);
    91.     // Convert the Indexed Labels back to the original Labels based on the trained predictions
    92.     val labelConverter = new IndexToString()
    93.                             .setInputCol("prediction")
    94.                             .setOutputCol("predicted_label")
    95.                             .setLabels(labelIndexer.labels);
    97.     // Chain the Indexers and Decision Tree Model to form a Pipeline
    98.     val pipeline = new Pipeline()
    99.                     .setStages(Array(labelIndexer, decisionTreeModel, labelConverter));
    101.     // Run the Indexers and Train the Model on the Training Data
    102.     return pipeline.fit(trainingDF);
    104.   }
    106.   /**
    107.    * Build a Random Forest Classifier
    108.    * Train a Random Forest Model by supplying the training dataset that includes the label and feature vector columns
    109.    *
    110.    * @param featuresDF The full DataFrame containing the labels and feature vectors
    111.    * @param trainingDF The training split DataFrame to be used to train the Model
    112.    * @param labelColumn The name of the column containing the labels
    113.    * @param featuresColumn The name of the column containing the scaled feature vectors
    114.    * @return The PipelineModel containing our trained random forest model
    115.    *
    116.    */
    118.   def trainRandomForestModel(featurizedDF:Dataset[Row], trainingDF:Dataset[Row], labelColumn:String,
    119.       featuresColumn:String): PipelineModel = {
    121.     // Index the Labels
    122.     val labelIndexer = new StringIndexer()
    123.                           .setInputCol(labelColumn)
    124.                           .setOutputCol("indexed_label")
    125.                           .fit(featurizedDF);
    127.     // Define a Random Forest model
    128.     val randomForestModel = new RandomForestClassifier()
    129.                               .setLabelCol("indexed_label")
    130.                               .setFeaturesCol(featuresColumn)
    131.                               .setNumTrees(numTrees);
    133.     // Convert the Indexed Labels back to the original Labels based on the trained predictions
    134.     val labelConverter = new IndexToString()
    135.                             .setInputCol("prediction")
    136.                             .setOutputCol("predicted_label")
    137.                             .setLabels(labelIndexer.labels);
    139.     // Chain the Indexers and Random Forest Model to form a Pipeline
    140.     val pipeline = new Pipeline()
    141.                       .setStages(Array(labelIndexer, randomForestModel, labelConverter));
    143.     // Run the Indexers and Train the Model on the Training Data
    144.     return pipeline.fit(trainingDF);
    146.   }
    148.   /**
    149.    * Generate Multi-class Metrics
    150.    * Generate multi-class metrics given a predictions dataframe containing prediction and indexed label double columns.
    151.    * Such metrics allow us to generate classification matrices, false and true positive rates etc.
    152.    *
    153.    * @param predictionsDF A DataFrame containing predictions and indexed labels
    154.    * @param predictionColumn The name of the column containing the predictions [Double]
    155.    * @param indexedLabelColumn The name of the column containing the indexed labels [Double]
    156.    * @return A MulticlassMetrics object that can be used to output model metrics
    157.    *
    158.    */
    160.   def generateMulticlassMetrics(predictionsDF:Dataset[Row], predictionColumn:String, indexedLabelColumn:String): MulticlassMetrics = {
    162.     val predictionAndLabels = predictionsDF.select(predictionColumn, indexedLabelColumn).rdd.map{
    163.       case Row(predictionColumn: Double, indexedLabelColumn:Double) => (predictionColumn, indexedLabelColumn);
    164.     };
    165.     return new MulticlassMetrics(predictionAndLabels);
    167.   }
    169. }

    训练决策树分类器
    现在让我们返回主应用程序并通过执行以下操作完成它：

    • 从我们的滤波Lemmas序列生成缩放的特征向量
    • 随机将我们的语料库分成训练（70％）和测试（30％）数据集
    • 在训练数据集上训练决策树分类器估计器
    • 将生成的决策树分类器模型应用于测试数据集，允许我们生成有关模型准确性的指标
    • 将我们的决策树分类器模型保存到HDFS，以便稍后在我们的实时基于事件的流数据管道中使用
    1. package io.keisan.knowledgebase.spark.mllib;
    3. import edu.stanford.nlp.pipeline.StanfordCoreNLP;
    5. import java.util.Properties;
    7. import org.apache.spark.SparkConf;
    8. import org.apache.spark.SparkContext;
    9. import org.apache.spark.ml.evaluation.MulticlassClassificationEvaluator;
    10. import org.apache.spark.sql.Dataset;
    11. import org.apache.spark.sql.functions._;
    12. import org.apache.spark.sql.types._;
    13. import org.apache.spark.sql.Row;
    14. import org.apache.spark.sql.SparkSession;
    15. import org.apache.spark.sql.SparkSession.Builder;
    17. import scala.collection.JavaConversions._;
    19. /**
    20.  * Tweet Sentiment Decision Tree Classifier
    21.  * Train a Decision Tree Classifier on a collection of pre-labelled tweets about airlines
    22.  *
    23.  * @author jillur.quddus
    24.  * @version 0.0.1
    25.  */
    27. object TrainDecisionTreeClassifier {
    29.   def main(args: Array[String]) = {
    31.     /********************************************************************
    32.      * SPARK CONTEXT
    33.      ********************************************************************/
    35.     // Create the Spark Context
    36.     val conf = new SparkConf()
    37.       .setAppName("Sentiment Models")
    38.       .setMaster("spark://<Spark Master Hostname>:7077");
    40.     val sc = new SparkContext(conf);
    41.     val sparkSession = SparkSession.builder().getOrCreate();
    42.     import sparkSession.implicits._;
    44.     /********************************************************************
    45.      * INGEST THE CORPUS
    46.      ********************************************************************/
    48.     // Define the CSV Dataset Schema
    49.     val schema = new StructType(Array(
    50.                     StructField("unit_id", LongType, true),
    51.                     StructField("golden", BooleanType, true),
    52.                     StructField("unit_state", StringType, true),
    53.                     StructField("trusted_judgements", IntegerType, true),
    54.                     StructField("last_judgement_at", StringType, true),
    55.                     StructField("airline_sentiment", StringType, true),
    56.                     StructField("airline_sentiment_confidence", DoubleType, true),
    57.                     StructField("negative_reason", StringType, true),
    58.                     StructField("negative_reason_confidence", DoubleType, true),
    59.                     StructField("airline", StringType, true),
    60.                     StructField("airline_sentiment_gold", StringType, true),
    61.                     StructField("name", StringType, true),
    62.                     StructField("negative_reason_gold", StringType, true),
    63.                     StructField("retweet_count", IntegerType, true),
    64.                     StructField("text", StringType, true),
    65.                     StructField("tweet_coordinates", StringType, true),
    66.                     StructField("tweet_created", StringType, true),
    67.                     StructField("tweet_id", StringType, true),
    68.                     StructField("tweet_location", StringType, true),
    69.                     StructField("user_timezone", StringType, true)
    70.                  ));
    72.     // Read the CSV Dataset, keeping only those columns that we need to build our model
    73.     var tweetsDF = SparkSession.builder().getOrCreate().read
    74.                       .format("csv")
    75.                       .option("header", true)
    76.                       .option("delimiter", ",")
    77.                       .option("mode", "DROPMALFORMED")
    78.                       .schema(schema)
    79.                       .load("hdfs://<Namenode Hostname>:9000/keisan/knowledgebase/spark/mllib/raw/Airline-Sentiment-2-w-AA.csv")
    80.                       .select("airline_sentiment", "text");
    82.     /********************************************************************
    83.      * LABEL THE DATA
    84.      ********************************************************************/
    86.     // We are interested in detecting tweets with negative sentiment. Let us create a new column whereby
    87.     // if the sentiment is negative, this new column is TRUE (Positive Outcome), and FALSE (Negative Outcome)
    88.     // in all other cases
    90.     tweetsDF = tweetsDF.withColumn("negative_sentiment_label", when(tweetsDF("airline_sentiment") === "negative", lit("true")).otherwise(lit("false")))
    91.                 .select("text", "negative_sentiment_label");
    93.     /********************************************************************
    94.      * APPLY THE PRE-PROCESSING PIPELINE
    95.      ********************************************************************/
    97.     // Let us now perform some simple pre-processing including converting the text column to lowercase
    98.     // and removing all non-alphanumeric characters
    100.     val lowercasedDF = PreProcessorUtils.lowercaseRemovePunctuation(tweetsDF, "text");
    102.     // Lemmatize the text to generate a sequence of Lemmas using the Stanford NLP Library
    103.     // By using mapPartitions, we create the Stanford NLP Pipeline once per partition rather than once per RDD entry
    105.     val lemmatizedDF = lowercasedDF.select("text", "negative_sentiment_label").rdd.mapPartitions(p => {
    107.       // Define the NLP Lemmatizer Pipeline once per partition
    108.       val props = new Properties();
    109.       props.put("annotators", "tokenize, ssplit, pos, lemma");
    110.       val pipeline = new StanfordCoreNLP(props);
    112.       // Lemmatize the text and preserve the Negative Sentiment Label
    113.       p.map{
    114.         case Row(text: String, negative_sentiment_label:String) => (PreProcessorUtils.lemmatizeText(text, pipeline), negative_sentiment_label);
    115.       };
    117.     }).toDF("lemmas", "negative_sentiment_label");
    119.     // Remove Stop Words from the sequence of Lemmas
    120.     val stopWordsRemovedDF = PreProcessorUtils.stopWordRemover(lemmatizedDF, "lemmas", "filtered_lemmas")
    121.                                 .where(size(col("filtered_lemmas")) > 1);
    123.     /********************************************************************
    124.      * SCALED FEATURE VECTOR
    125.      ********************************************************************/
    127.     // Generate the Scaled Feature Vectors
    128.     val featurizedDF = ModelUtils.tfidf(stopWordsRemovedDF, "filtered_lemmas", "features");
    130.     /********************************************************************
    131.      * TRAIN AND EVALUATE A DECISION TREE CLASSIFIER
    132.      ********************************************************************/
    134.     // Split the data into Training and Test Datasets
    135.     val Array(trainingDF, testDF) = featurizedDF.randomSplit(Array(0.7, 0.3))
    137.     // Train a Decision Tree Model using the Training Dataset
    138.     val decisionTreeModel = ModelUtils.trainDecisionTreeModel(featurizedDF, trainingDF, "negative_sentiment_label", "features");
    140.     // Apply the Decision Tree Training Model to the Test Dataset
    141.     val decisionTreePredictions = decisionTreeModel.transform(testDF);
    142.     decisionTreePredictions.select("negative_sentiment_label", "predicted_label", "filtered_lemmas", "features").show(false);
    144.     // Compute the accuracy of the Decision Tree Training Model on the Test Dataset
    145.     val decisionTreeEvaluator = new MulticlassClassificationEvaluator()
    146.                                   .setLabelCol("indexed_label")
    147.                                   .setPredictionCol("prediction")
    148.                                   .setMetricName("accuracy");
    149.     val decisionTreeAccuracy = decisionTreeEvaluator.evaluate(decisionTreePredictions);
    150.     println("Decision Tree Test Accuracy Rate = " + decisionTreeAccuracy);
    151.     println("Decision Tree Test Error Rate = " + (1.0 - decisionTreeAccuracy));
    153.     // Generate a Classification Matrix
    154.     val metrics = ModelUtils.generateMulticlassMetrics(decisionTreePredictions, "prediction", "indexed_label");
    155.     println(metrics.confusionMatrix);
    157.     // Generate Label Accuracy Metrics
    158.     val labelMetrics = metrics.labels;
    159.     labelMetrics.foreach { l =>
    160.       println(s"False Positive Rate ($l) = " + metrics.falsePositiveRate(l));
    161.     }
    163.     /********************************************************************
    164.      * SAVE THE DECISION TREE CLASSIFIER FOR REAL-TIME STREAMING
    165.      ********************************************************************/
    167.     decisionTreeModel.save("hdfs://<Namenode Hostname>:9000/keisan/knowledgebase/spark/mllib/models/decisionTreeClassifier");
    169.   }
    171. }

    多级度量标准
    上面的应用程序将打印一些关于我们训练的决策树分类器模型在测试数据集上的准确性的指标。确切的数字可能与您的略有不同，因为我们随机分割语料库，但通常您应该看到以下结果：

    • 测试准确率：71％
    • 测试错误率：29％，

    和以下分类矩阵：

    2285.0 293.0
    945.0 681.0

    该分类矩阵告诉我们，在4,204次测试观察中，经过训练的决策树分类器模型得到2,285个预测正确，从而预测负面情绪，测试推文实际上是负面情绪（真实肯定）。该模型得到了进一步的681个预测正确，从而预测了积极的情绪，测试推文实际上是积极的情绪（真阴性）。因此，在总共4,204次测试观察中，该模型得到2,966个预测正确 - 准确率为71％。
    然而，通过遵循分类矩阵中的相反对角线，模型得到293个预测不正确，从而预测正面情绪，但测试推文实际上是负面情绪（假阴性）。该模型得到了进一步的945预测不正确，因此它预测负面情绪，但测试推文实际上是积极的情绪（误报）。因此，在总共4,204次测试观察中，该模型得出1,238个预测错误 - 错误率为29％。
    因此，我们的决策树分类器在71％的时间内正确地预测了推文的情绪。我们现在将我们训练有素的模型应用于我们基于事件的实时流媒体数据管道，以尝试预测航空公司推文创建时的情绪！
    Apache Kafka 生产者
    在我的上一篇文章中，我们更新了我们的数据管道，使用Apache Kafka Producer在Twitter4j包装器中使用Twitter的Hosebird客户端收集推文，其中推文使用Avro库和Bijection进行序列化。如果您运行该Producer，您应该观察发布到Kafka主题的推文。
    Apache Spark消费者流应用程序
    我们现在要创建一个Apache Spark Consumer Streaming Application，它将使用来自Kafka主题的Avro推文。然后，我们将扩展我们的高级实时数据处理流程，方法是将这些推文传递给上述相同的预处理管道，然后再对其应用我们经过训练的决策树分类器，尝试近乎实时地对其潜在情绪进行分类。 ！

    image.jpeg

    我将在我们上面训练决策树分类器的同一个项目中编写Spark Consumer Streaming Application。或者，您可以为Consumer Streaming Application创建一个新项目 - 这取决于您。由于我将使用相同的项目，我需要更新我的Maven POM以包含一些额外的依赖项 - 即Apache Avro，Apache Spark Streaming（已提供），Apache Kafka（请注意，我将使用与之兼容的0.8.2.2版本）我从上一篇文章中得到了 Spark 2.1.0 Streaming，Bijection和我的Kafka Twitter Producer应用程序（在这里我定义了Avro Tweet Schema，当我从Kafka主题消费并解析消息时我将再次需要它）：

    1. <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    2.   xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    3.     <modelVersion>4.0.0</modelVersion>
    5.     <!-- Project Information -->
    6.     <groupId>io.keisan.knowledgebase.spark.mllib</groupId>
    7.     <artifactId>keisan-spark-mllib</artifactId>
    8.     <version>0.0.1-SNAPSHOT</version>
    9.     <packaging>jar</packaging>
    11.     <!-- Project Information -->
    12.     <name>Example Spark Predictive Models</name>
    13.     <description>Keisan Knowledgebase Spark MLlib Project - Example Spark Machine Learning Pipelines</description>
    14.     <url>https://www.keisan.io/knowledgebase/real-time-ml-pipeline-spark</url>
    15.     <organization>
    16.         <name>Keisan Ltd</name>
    17.         <url>https://www.keisan.io</url>
    18.     </organization>
    19.     <developers>
    20.         <developer>
    21.             <id>jillur.quddus</id>
    22.             <name>Jillur Quddus</name>
    23.             <email>contactus@keisan.io</email>
    24.             <url>https://www.keisan.io</url>
    25.             <organization>Keisan Ltd</organization>
    26.             <organizationUrl>https://www.keisan.io</organizationUrl>
    27.             <roles>
    28.                 <role>Lead Engineer</role>
    29.                 <role>Data Scientist</role>
    30.             </roles>
    31.             <timezone>Europe/London</timezone>
    32.         </developer>
    33.     </developers>
    35.     <!-- Properties -->
    36.     <properties>
    37.         <apache.avro.version>1.8.1</apache.avro.version>
    38.         <apache.kafka.2.11.version>0.8.2.2</apache.kafka.2.11.version>
    39.         <apache.spark.core.2.11.version>2.1.0</apache.spark.core.2.11.version>
    40.         <apache.spark.mllib.2.11.version>2.1.0</apache.spark.mllib.2.11.version>
    41.         <apache.spark.sql.2.11.version>2.1.0</apache.spark.sql.2.11.version>
    42.         <apache.spark.streaming.2.11.version>2.1.0</apache.spark.streaming.2.11.version>
    43.         <apache.spark.streaming.kafka-0-8_2.11.version>2.1.0</apache.spark.streaming.kafka-0-8_2.11.version>
    44.         <bijection.avro.2.11.version>0.9.5</bijection.avro.2.11.version>
    45.         <jdk.version>1.8</jdk.version>
    46.         <keisan.kafka.producers.version>0.0.1-SNAPSHOT</keisan.kafka.producers.version>
    47.         <maven.plugins.maven-assembly-plugin.version>3.0.0</maven.plugins.maven-assembly-plugin.version>
    48.         <maven.plugins.maven-compiler-plugin.version>3.6.1</maven.plugins.maven-compiler-plugin.version>
    49.         <output.directory>/keisan/knowledgebase/spark/mllib/jars</output.directory>
    50.         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    51.         <scala.library.version>2.11.8</scala.library.version>
    52.         <scala.maven.plugin.version>3.2.2</scala.maven.plugin.version>
    53.         <stanford.corenlp.version>3.7.0</stanford.corenlp.version>
    54.     </properties>
    56.     <!-- Dependencies -->
    57.     <dependencies>
    59.         <!-- Apache Avro -->
    60.         <dependency>
    61.             <groupId>org.apache.avro</groupId>
    62.             <artifactId>avro</artifactId>
    63.             <version>${apache.avro.version}</version>
    64.         </dependency>
    66.         <!-- Apache Kafka -->
    67.         <dependency>
    68.             <groupId>org.apache.kafka</groupId>
    69.             <artifactId>kafka_2.11</artifactId>
    70.             <version>${apache.kafka.2.11.version}</version>
    71.         </dependency>
    73.         <!-- Apache Spark -->
    74.         <dependency>
    75.             <groupId>org.apache.spark</groupId>
    76.             <artifactId>spark-core_2.11</artifactId>
    77.             <version>${apache.spark.core.2.11.version}</version>
    78.             <scope>provided</scope>
    79.         </dependency>
    80.         <dependency>
    81.             <groupId>org.apache.spark</groupId>
    82.             <artifactId>spark-mllib_2.11</artifactId>
    83.             <version>${apache.spark.mllib.2.11.version}</version>
    84.             <scope>provided</scope>
    85.         </dependency>
    86.         <dependency>
    87.             <groupId>org.apache.spark</groupId>
    88.             <artifactId>spark-sql_2.11</artifactId>
    89.             <version>${apache.spark.sql.2.11.version}</version>
    90.             <scope>provided</scope>
    91.         </dependency>
    92.         <dependency>
    93.             <groupId>org.apache.spark</groupId>
    94.             <artifactId>spark-streaming_2.11</artifactId>
    95.             <version>${apache.spark.streaming.2.11.version}</version>
    96.             <scope>provided</scope>
    97.         </dependency>
    98.         <dependency>
    99.             <groupId>org.apache.spark</groupId>
    100.             <artifactId>spark-streaming-kafka-0-8_2.11</artifactId>
    101.             <version>${apache.spark.streaming.kafka-0-8_2.11.version}</version>
    102.         </dependency>
    104.         <!-- Bijection -->
    105.         <dependency>
    106.             <groupId>com.twitter</groupId>
    107.             <artifactId>bijection-avro_2.11</artifactId>
    108.             <version>${bijection.avro.2.11.version}</version>
    109.         </dependency>
    111.         <!-- Keisan Kafka Producers -->
    112.         <dependency>
    113.             <groupId>io.keisan.knowledgebase.kafka.producers</groupId>
    114.             <artifactId>keisan-kafka-producers</artifactId>
    115.             <version>${keisan.kafka.producers.version}</version>
    116.         </dependency>
    118.         <!-- Scala -->
    119.         <dependency>
    120.             <groupId>org.scala-lang</groupId>
    121.             <artifactId>scala-library</artifactId>
    122.             <version>${scala.library.version}</version>
    123.         </dependency>
    125.         <!-- Stanford NLP -->
    126.         <dependency>
    127.             <groupId>edu.stanford.nlp</groupId>
    128.             <artifactId>stanford-corenlp</artifactId>
    129.             <version>${stanford.corenlp.version}</version>
    130.         </dependency>
    131.         <dependency>
    132.             <groupId>edu.stanford.nlp</groupId>
    133.             <artifactId>stanford-corenlp</artifactId>
    134.             <version>${stanford.corenlp.version}</version>
    135.             <classifier>models</classifier>
    136.         </dependency>
    138.     </dependencies>
    140.     <!-- Project Builder -->
    141.     <build>
    143.         <!-- Plugins -->
    144.         <plugins>
    146.             <!-- Maven Compiler: Compile the Sources of the Project -->
    147.             <plugin>
    148.                 <groupId>org.apache.maven.plugins</groupId>
    149.                 <artifactId>maven-compiler-plugin</artifactId>
    150.                 <version>${maven.plugins.maven-compiler-plugin.version}</version>
    151.                 <configuration>
    152.                     <source>${jdk.version}</source>
    153.                     <target>${jdk.version}</target>
    154.                 </configuration>
    155.             </plugin>
    157.             <!-- Maven Assembly: Aggregate project output with its dependencies -->
    158.             <plugin>
    159.                 <groupId>org.apache.maven.plugins</groupId>
    160.                 <artifactId>maven-assembly-plugin</artifactId>
    161.                 <version>${maven.plugins.maven-assembly-plugin.version}</version>
    162.                 <configuration>
    164.                     <!-- Final JAR Filename -->
    165.                     <finalName>keisan-spark-mllib-${project.version}</finalName>
    166.                     <appendAssemblyId>false</appendAssemblyId>
    168.                     <!-- Include all Project Dependencies -->
    169.                     <descriptorRefs>
    170.                         <descriptorRef>jar-with-dependencies</descriptorRef>
    171.                     </descriptorRefs>
    173.                     <!-- JAR with dependencies Output Target Directory -->
    174.                     <outputDirectory>${output.directory}</outputDirectory>
    176.                 </configuration>
    177.                 <executions>
    179.                     <!-- Bind the assembly to the package phase -->
    180.                     <execution>
    181.                         <id>make-assembly</id>
    182.                         <phase>package</phase>
    183.                         <goals>
    184.                             <goal>single</goal>
    185.                         </goals>
    186.                     </execution>
    188.                 </executions>
    190.             </plugin>
    192.             <!-- Scala Maven Plugin -->
    193.             <plugin>
    194.                 <groupId>net.alchim31.maven</groupId>
    195.                 <artifactId>scala-maven-plugin</artifactId>
    196.                 <version>${scala.maven.plugin.version}</version>
    197.             </plugin>
    199.         </plugins>
    201.     </build>
    203. </project>

    Spark Streaming Context
    我将在Scala中编写Spark消费者流应用程序，但Java等效项应该在结构上类似，尽管更冗长。首先要做的是创建Spark Streaming Context。我将把正常的Spark Context传递给Spark Streaming Context Constructors之一来创建Streaming Context，在我的例子中使用5秒的批处理间隔。

    1. // Spark Streaming Context
    2. val conf = new SparkConf()
    3.     .setAppName("Streaming Twitter Sentiment Classifier")
    4.     .setMaster("spark://<Spark Master Hostname>:7077");
    5. val sc = new SparkContext(conf);
    6. val ssc = new StreamingContext(sc, Seconds(5));
    7. val sparkSession = SparkSession.builder().getOrCreate();
    8. import sparkSession.implicits._;

    加载管道模型
    第二步是将我们先前持久化的训练决策树分类器流水线模型加载到HDFS。要做到这一点很简单：

    1. // Load the Trained Decision Tree Classifier
    2. val decisionTreeModel = PipelineModel.read.load("hdfs://<Namenode Hostname>:9000/keisan/knowledgebase/spark/mllib/models/decisionTreeClassifier");

    实时流媒体情感分析
    我们现在准备编写我们的主要Consumer Streaming应用程序。我们的Spark Streaming应用程序将执行以下操作：

    • Kafka直接流 - 我们将使用默认解码器定期向Kafka查询每个主题和分区中的最新偏移量，以返回表示我们的Avro推文的原始字节数组，从而产生离散化的流。
    • Bijection - 然后我们将通过使用Bijection和我的Kafka Twitter Producer中定义的Tweet Schema反转原始字节数组来开始处理离散化的流，从而产生一个RDD of Rows，然后我们将转换为包含我们解析的tweet字段的DataFrame，包括推文的核心文本。
    • 预处理 - 接下来，我们将对我们在训练决策树分类器时应用的这些DataFrame应用相同的预处理管道，即小写文本，删除无意义的标点符号，使用斯坦福的Core NLP库对文本进行词典化，最后删除从由此产生的引理序列中停止。
    • 缩放特征向量 - 接下来，我们将使用TF-IDF从这些过滤的词条序列生成缩放特征向量。
    • 决策树分类器预测 - 最后，我们将这些缩放的特征向量传递给我们之前创建和加载的训练过的决策树分类器，以预测推文中的推文是负面还是非负面！
    1. package io.keisan.knowledgebase.spark.mllib;
    3. import com.twitter.bijection.Injection;
    4. import com.twitter.bijection.avro.GenericAvroCodecs;
    6. import edu.stanford.nlp.pipeline.StanfordCoreNLP;
    8. import io.keisan.knowledgebase.kafka.producers.TwitterProducer;
    10. import java.util.HashMap;
    11. import java.util.Properties;
    13. import kafka.serializer.DefaultDecoder;
    14. import kafka.serializer.StringDecoder;
    16. import org.apache.avro.Schema;
    17. import org.apache.avro.generic.GenericRecord;
    18. import org.apache.spark.ml.PipelineModel;
    19. import org.apache.spark.rdd.RDD;
    20. import org.apache.spark.SparkConf;
    21. import org.apache.spark.SparkContext;
    22. import org.apache.spark.sql.Dataset;
    23. import org.apache.spark.sql.Row;
    24. import org.apache.spark.sql.SparkSession;
    25. import org.apache.spark.sql.functions._;
    26. import org.apache.spark.sql.types._;
    27. import org.apache.spark.streaming.Seconds;
    28. import org.apache.spark.streaming.StreamingContext;
    29. import org.apache.spark.streaming.kafka.KafkaUtils;
    31. import scala.collection.JavaConversions._;
    33. /**
    34.  * Streaming Twitter Sentiment Classifier
    35.  * Consume and deserialise Avro Tweets from Kafka Topics. Apply the same pre-processing
    36.  * pipeline to the stream of tweets as our Decision Tree Classifier trainer and generate
    37.  * scaled Feature Vectors. Finally apply the trained Decision Tree Classifier on the
    38.  * stream of tweets to predict their sentiment in near real-time.
    39.  *
    40.  * Usage: KafkaStreamingSentimentClassifier <brokers> <topics>
    41.  *  brokers: List of one or more Kafka Brokers
    42.  *  topics: List of one or more Kafka Topics to consume from
    43.  *
    44.  * @author jillur.quddus
    45.  * @version 0.0.1
    46.  */
    48. object KafkaStreamingSentimentClassifier {
    50.   // Spark Streaming Context
    51.   val conf = new SparkConf()
    52.       .setAppName("Streaming Twitter Sentiment Classifier")
    53.       .setMaster("spark://<Spark Master Hostname>:7077");
    54.   val sc = new SparkContext(conf);
    55.   val ssc = new StreamingContext(sc, Seconds(5));
    56.   val sparkSession = SparkSession.builder().getOrCreate();
    57.   import sparkSession.implicits._;
    59.     // Load the Trained Decision Tree Classifier
    60.     val decisionTreeModel = PipelineModel.read.load("hdfs://<Namenode Hostname>:9000/keisan/knowledgebase/spark/mllib/models/decisionTreeClassifier");
    62.     /**
    63.      * Consume and deserialise Avro messages from the Kafka Topics using a Direct Stream Approach.
    64.      * Pass the stream of tweets through the same pre-processing pipeline as the training dataset
    65.      * and generate scaled Feature Vectors. Pass the scaled Feature Vectors to the trained
    66.      * Decision Tree Classifier to predict the underlying sentiment.
    67.      *
    68.      * @param args List of Kafka Brokers and Topics from which to consume from
    69.      */
    71.   def main(args: Array[String]) = {
    73.     if ( args.length < 2 ) {
    74.             System.err.println("Usage: KafkaStreamingSentimentClassifier <brokers> <topics>");
    75.             System.exit(1);
    76.         }
    78.     val Array(brokers, topics) = args;
    80.     /********************************************************************
    81.      * KAFKA CONSUMER DSTREAM
    82.      ********************************************************************/
    84.     // Specify the Kafka Broker Options and set of Topics
    85.     val kafkaParameters = Map[String, String]("metadata.broker.list" -> brokers);
    86.     val topicSet = topics.split(",").toSet;
    88.     // Create an input DStream using KafkaUtils and the DefaultDecoder to provide the raw array of Bytes
    89.     val messages = KafkaUtils.createDirectStream[String, Array[Byte], StringDecoder, DefaultDecoder](
    90.         ssc, kafkaParameters, topicSet);
    92.     /********************************************************************
    93.      * DESERIALISE USING INVERSION, PRE-PROCESS AND CLASSIFY
    94.      ********************************************************************/
    96.     // Deserialise the Avro messages using Bijection and the Tweet Schema
    97.     messages
    98.       .map(message => {
    100.         val parser = new Schema.Parser();
    101.         val schema = parser.parse(TwitterProducer.TWEET_SCHEMA);
    102.         val recordInjection = GenericAvroCodecs.toBinary[GenericRecord](schema);
    103.         val record = recordInjection.invert(message._2).get;
    104.         Row(record.get("id").toString(), record.get("user_name").toString(), record.get("text").toString());
    106.       })
    107.       .foreachRDD(rdd => {
    109.         // Check that the RDD is not null - otherwise applying the predictive model will raise an IllegalStateException
    110.         if (rdd != null && !rdd.isEmpty()) {
    112.             // Convert the RDD of Rows to a DataFrame of Tweet ID, Username and Text
    113.             val tweetsDF = sparkSession.createDataFrame(rdd,
    114.                 new StructType().add("id", StringType)
    115.                   .add("user_name", StringType)
    116.                   .add("text", StringType));
    118.             /********************************************************************
    119.              * APPLY THE SAME PRE-PROCESSING PIPELINE TO THE REAL-TIME TWEETS
    120.              ********************************************************************/
    122.             // Lowercase and remove punctuation
    123.             val lowercasedDF = PreProcessorUtils.lowercaseRemovePunctuation(tweetsDF, "text");
    125.             // Lemmatize
    126.             val lemmatizedDF = lowercasedDF.select("user_name", "text").rdd.mapPartitions(p => {
    127.                 val props = new Properties();
    128.                 props.put("annotators", "tokenize, ssplit, pos, lemma");
    129.                 val pipeline = new StanfordCoreNLP(props);
    130.                 p.map{
    131.                     case Row(user_name:String, text: String) => (user_name, text, PreProcessorUtils.lemmatizeText(text, pipeline));
    132.                 };
    133.             }).toDF("user_name", "text", "lemmas");
    135.             // Remove Stop Words from the sequence of Lemmas
    136.             val stopWordsRemovedDF = PreProcessorUtils.stopWordRemover(lemmatizedDF, "lemmas", "filtered_lemmas")
    137.                                         .where(size(col("filtered_lemmas")) > 1);
    139.             /********************************************************************
    140.              * SCALED FEATURE VECTORS
    141.              ********************************************************************/
    143.             // Generate the Scaled Feature Vectors
    144.             val featurizedDF = ModelUtils.tfidf(stopWordsRemovedDF, "filtered_lemmas", "features");
    146.             /********************************************************************
    147.             * APPLY TRAINED DECISION TREE CLASSIFIER TO THE REAL-TIME TWEETS
    148.              ********************************************************************/
    150.             // Apply the Decision Tree Classifier to the scaled Feature Vectors
    151.             // Note that the Decision Tree Classifier was trained using a column called "features".
    152.             // A column named "features" MUST also be present on the new dataset in order to make
    153.             // predictions. If it is called anything else, an exception will be raised.
    154.             val predictions = decisionTreeModel.transform(featurizedDF);
    156.             // Output the original tweet's username, text and predicted label
    157.             predictions.select("user_name", "text", "predicted_label").show(false);
    159.         }
    161.       });
    163.     // Start the computation
    164.     ssc.start();
    166.     // Wait for the computation to terminate
    167.     ssc.awaitTermination();
    169.   }
    171. }

    我们现在可以运行我们的端到端机器学习管道！首先，我将稍微更新我的Kafka Twitter 生产者，根据英国的常见航空公司进行过滤，包括British_Airways，Ryanair和easyJet。假设我的Kafka集群在线，并且我的Kafka Producer正在运行，我可以使用Maven构建我的Spark Consumer Streaming Application并使用以下命令将其部署到我的Spark集群：

    1. # Deploy the Spark Consumer Streaming Application to the Spark Cluster
    2. bin/spark-submit --class io.keisan.knowledgebase.spark.mllib.KafkaStreamingSentimentClassifier --master spark://<Spark Master Hostname>:6066 --deploy-mode cluster /keisan/knowledgebase/spark/mllib/jars/keisan-spark-mllib-0.0.1-SNAPSHOT.jar <Kafka Broker Hostname>:9092 twitter

    假设一切顺利，如果您检查Spark驱动程序的标准输出，您应该看到Spark Dataframes正在打印，包含推文的用户名，文本和预测标签 - 请记住，在我们的案例中，真正的结果是推文是负面的情绪，而错误的结果是推文在情绪上是积极的。以下是一些使用我们的决策树分类器自动分类的推文：

    image.png

    正如您所看到的，我们的决策树分类器并不完全完美，但在大多数情况下，它已正确预测了推文的潜在情绪！航空公司和其他组织可以使用这样的预测模型来告知他们如何近乎实时地改进他们的服务和产品。
    在本文中，我们成功开发了一种近实时，高吞吐量，可靠且容错的机器学习管道，能够在流式传输和接收时对基于事件的数据进行准确预测。在以后的文章中，我将探讨其他类型的监督和非监督算法，以及如何将它们合并到实时管道中。