拟合管道和处理数据答案

【问题标题】：Fitting pipeline and processing the data拟合管道和处理数据
【发布时间】：2017-02-21 02:22:17
【问题描述】：

我有一个包含文本的文件。我想要做的是使用管道对文本进行标记，删除停用词并生成 2-gram。

到目前为止我做了什么：

第 1 步：读取文件

val data = sparkSession.read.text("data.txt").toDF("text")

第 2 步：构建管道

val pipe1 = new Tokenizer().setInputCol("text").setOutputCol("words")
val pipe2 = new StopWordsRemover().setInputCol("words").setOutputCol("filtered")
val pipe3 = new NGram().setN(2).setInputCol("filtered").setOutputCol("ngrams")

val pipeline = new Pipeline().setStages(Array(pipe1, pipe2, pipe3))
val model = pipeline.fit(data)

我知道pipeline.fit(data) 会生成PipelineModel，但我不知道如何使用PipelineModel。

任何帮助将不胜感激。

【问题讨论】：

标签： scala apache-spark pipeline

【解决方案1】：

当您运行val model = pipeline.fit(data) 代码时，所有Estimator 阶段（即：分类、回归、聚类等机器学习任务）都适合数据并创建Transformer 阶段。您只有 Transformer 阶段，因为您正在此管道中创建功能。

为了执行您的模型，现在只包含Transformer 阶段，您需要运行val results = model.transform(data)。这将针对您的数据框执行每个 Transformer 阶段。因此，在model.transform(data) 进程结束时，您将拥有一个由原始行、Tokenizer 输出、StopWordsRemover 输出以及最后的 NGram 结果组成的数据帧。

在特征创建完成后发现前 5 个 ngram 可以通过 SparkSQL 查询来执行。先炸开ngram列，然后count groupby ngrams，按计数列降序排序，然后执行show(5)。或者，您可以使用"LIMIT 5 方法代替show(5)。

顺便说一句，您可能应该将您的对象名称更改为非标准类名称。否则你会得到一个模棱两可的范围错误。

代码：

import org.apache.spark.sql.SparkSession
import org.apache.spark.ml.feature.Tokenizer
import org.apache.spark.sql.SparkSession._
import org.apache.spark.sql.functions._
import org.apache.spark.ml.feature.NGram
import org.apache.spark.ml.feature.StopWordsRemover
import org.apache.spark.ml.{Pipeline, PipelineModel}

object NGramPipeline {
    def main() {
        val sparkSession = SparkSession.builder.appName("NGram Pipeline").getOrCreate()

        val sc = sparkSession.sparkContext

        val data = sparkSession.read.text("quangle.txt").toDF("text")

        val pipe1 = new Tokenizer().setInputCol("text").setOutputCol("words")
        val pipe2 = new StopWordsRemover().setInputCol("words").setOutputCol("filtered")
        val pipe3 = new NGram().setN(2).setInputCol("filtered").setOutputCol("ngrams")

        val pipeline = new Pipeline().setStages(Array(pipe1, pipe2, pipe3))
        val model = pipeline.fit(data)

        val results = model.transform(data)

        val explodedNGrams = results.withColumn("explNGrams", explode($"ngrams"))
        explodedNGrams.groupBy("explNGrams").agg(count("*") as "ngramCount").orderBy(desc("ngramCount")).show(10,false)

    }
}
NGramPipeline.main()

输出：

+-----------------+----------+
|explNGrams       |ngramCount|
+-----------------+----------+
|quangle wangle   |9         |
|wangle quee.     |4         |
|'mr. quangle     |3         |
|said, --         |2         |
|wangle said      |2         |
|crumpetty tree   |2         |
|crumpetty tree,  |2         |
|quangle wangle,  |2         |
|crumpetty tree,--|2         |
|blue babboon,    |2         |
+-----------------+----------+
only showing top 10 rows

请注意，存在导致行重复的语法（逗号、破折号等）。在执行 ngram 时，过滤我们的语法通常是一个好主意。您通常可以使用正则表达式来执行此操作。

【讨论】：

关于我所说的“变形金刚”的具体细节，您可以阅读管道文档的“工作原理”部分。 spark.apache.org/docs/latest/ml-pipeline.html#how-it-works
感谢您的回复。您能否还解释一下我应该怎么做才能获得 5 个最常见的 2 克？在您添加的代码中，显示了前 5 个 2-gram（而不是最常见的 5 个）。
当然，一旦创建了 ngram 功能，这只是 sql 查询。它已添加到示例中。
这些是 ngram 对，所以一对是 (quangle wangle, qangle quee)，另一对是 (quangle wangle, wangle said)。
那么您正在寻找爆炸功能。对不起，误会了。我将使用爆炸更新示例。