Montage der Pipeline und Verarbeitung der Daten

Question

Ich habe eine Datei, die Text enthält. Was ich tun möchte, ist eine Pipeline zu verwenden, um den Text zu entwerten, die Stoppwörter zu entfernen und 2 Gramm zu produzieren.

Was ich bisher getan habe:

Schritt 1: Lesen Sie die Datei

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

Schritt 2: Erstellen Sie die Pipeline

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) 

Ich weiß, dass pipeline.fit(data) produziert eine PipelineModel aber ich weiß nicht, wie man eine PipelineModel verwendet.

Jede Hilfe würde sehr geschätzt werden.

Answer 1

Wenn Sie führen Sie den Code val model = pipeline.fit(data) alle Estimator Stufen (dh: Machine Learning Aufgaben wie Klassifikationen, Regressionen, Clustering, etc.), um die Daten angepasst sind und eine Transformer Bühne erstellt wird. Sie haben nur Transformer Stages, da Sie Feature-Erstellung in dieser Pipeline machen.

Um Ihr Modell auszuführen, das nur aus Transformer Stufen besteht, müssen Sie val results = model.transform(data) ausführen. Dies führt jede Transformer Bühne gegen Ihren Datenrahmen. So haben Sie am Ende des model.transform(data) Prozesses einen Datenrahmen, der aus den ursprünglichen Zeilen, der Tokenizer-Ausgabe, der StopWordsRemover-Ausgabe und schließlich den NGram-Ergebnissen besteht.

Die Ermittlung der obersten 5 Ngrams nach Abschluss der Feature-Erstellung kann über eine SparkSQL-Abfrage durchgeführt werden. Zuerst die ngram-Spalte auflösen, dann groupby ngrams zählen, absteigend nach der gezählten Spalte sortieren und dann eine show(5) ausführen. Alternativ können Sie auch eine "LIMIT 5-Methode anstelle von show(5) verwenden.

Nebenbei sollten Sie wahrscheinlich Ihren Objektnamen in etwas ändern, das kein Standardklassenname ist. Sonst werden Sie einen zweideutigen Scope-Fehler bekommen.

CODE:

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() 

OUTPUT:

+-----------------+----------+ 
|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 

bemerken, dass es folgende Syntax (Komma, Striche usw.), die Linien verursacht werden dupliziert. Wenn Sie Ngrams ausführen, ist es oft eine gute Idee, unsere Syntax zu filtern. Sie können dies normalerweise mit einer Regex tun.