Разделить строки на слова в искре scala

Question

У меня большой текстовый файл, который содержит идентификатор документа в виде столбца, и текст соответствует каждому идентификатору документа. Мне нужно получить количество каждого слова в каждом документе. Я использую спарк scala с использованием блоков данных.

Данные >>> https://drive.google.com/open?id=1g4YYALk3nArN8bX2uFS70IpbdSf_Efqj

Сначала я импортировал данные следующим образом:

 import org.apache.spark.sql.types._
    val schema = new StructType().add("DocumentID", LongType, true).add("Description", StringType, true)
    val data = spark.read.format("csv").option("delimiter", "\t").schema(schema).load("/data1.txt")
    data.show(10)

    +----------+--------------------+
    |DocumentID|         Description|
    +----------+--------------------+
    |  23890098|Shlykov, a hard-w...|
    |  31186339|The nation of Pan...|
    |  20663735|Poovalli Induchoo...|
    |   2231378|The Lemon Drop Ki...|
    |    595909|Seventh-day Adven...|
    |   5272176|The president is ...|
    |   1952976|{{plot}} The film...|
    |  24225279|The story begins ...|
    |   2462689|Infuriated at bei...|
    |  20532852|A line of people ...|
    +----------+--------------------+

Далее мне нужно разбить каждую строку на получить количество слов, и мой желаемый результат должен быть примерно таким,

Я устал после кода,

val columns = data.columns.map(col) :+
      (explode(col("Description")) as "token")
val unfoldedDocs = data.select(columns: _*)

Но, кажется, есть некоторые проблемы с что.

Кто-нибудь может мне помочь разобраться в этом?

Спасибо

Answer 1

Сначала split Описание данных для получения array, а затем explode массив для получения отдельных words как строки, связанные с DocumentID.

Example:

import org.apache.spark.sql.functions._
import org.apache.spark.sql.types._

//sampledata
val csvData: Dataset[String] = spark.sparkContext.parallelize(
  """
|23890098   Shlykov, a hard-working taxi driver and Lyosha, a saxophonist, develop a bizarre love-hate relationship, and despite their prejudices, realize they aren't so different after all.
  """.stripMargin.lines.toList).toDS()

val schema = new StructType().add("DocumentID", LongType, true).add("Description", StringType, true)

//read the dataset
val data=spark.read.option("delimiter", "\t").schema(schema).csv(csvData)

data.withColumn("Description",split(col("Description")," ")).//split description to get array of words
    withColumn("token", explode(col("Description"))). //explode on array and get tokens(each word as individual row with documentid)
    show(false)

//+----------+--------------------+-------------+
//|DocumentID|         Description|        token|
//+----------+--------------------+-------------+
//|  23890098|[Shlykov,, a, har...|     Shlykov,|
//|  23890098|[Shlykov,, a, har...|            a|
//|  23890098|[Shlykov,, a, har...| hard-working|
//|  23890098|[Shlykov,, a, har...|         taxi|
//|  23890098|[Shlykov,, a, har...|       driver|
//|  23890098|[Shlykov,, a, har...|          and|
//|  23890098|[Shlykov,, a, har...|      Lyosha,|
//|  23890098|[Shlykov,, a, har...|            a|
//|  23890098|[Shlykov,, a, har...| saxophonist,|
//|  23890098|[Shlykov,, a, har...|      develop|
//|  23890098|[Shlykov,, a, har...|            a|
//|  23890098|[Shlykov,, a, har...|      bizarre|
//|  23890098|[Shlykov,, a, har...|    love-hate|
//|  23890098|[Shlykov,, a, har...|relationship,|
//|  23890098|[Shlykov,, a, har...|          and|
//|  23890098|[Shlykov,, a, har...|      despite|
//|  23890098|[Shlykov,, a, har...|        their|
//|  23890098|[Shlykov,, a, har...|  prejudices,|
//|  23890098|[Shlykov,, a, har...|      realize|
//|  23890098|[Shlykov,, a, har...|         they|
//+----------+--------------------+-------------+
//only showing top 20 rows

Answer 2

Я думаю, что комбинация Tokenizer и explode может работать. Решение приведено ниже:

scala> val data = spark.read.format("csv").option("delimiter", "\t").schema(schema).load("plot_summaries.txt")
data: org.apache.spark.sql.DataFrame = [DocumentID: bigint, Description: string]

scala> data.show(1)
+----------+--------------------+
|DocumentID|         Description|
+----------+--------------------+
|  23890098|Shlykov, a hard-w...|
+----------+--------------------+
only showing top 1 row


scala> import org.apache.spark.sql.functions.explode
import org.apache.spark.sql.functions.explode

scala> import org.apache.spark.ml.feature.Tokenizer
import org.apache.spark.ml.feature.Tokenizer

scala> val tokenizer = new Tokenizer().setInputCol("Description").setOutputCol("Words")
tokenizer: org.apache.spark.ml.feature.Tokenizer = tok_80d1c6e72cbc

scala> val wordsData = tokenizer.transform(data)
wordsData: org.apache.spark.sql.DataFrame = [DocumentID: bigint, Description: string ... 1 more field]

scala> wordsData.show(1)
+----------+--------------------+--------------------+
|DocumentID|         Description|               Words|
+----------+--------------------+--------------------+
|  23890098|Shlykov, a hard-w...|[shlykov,, a, har...|
+----------+--------------------+--------------------+
only showing top 1 row


scala> val newWordsData = wordsData.drop("Description")
newWordsData: org.apache.spark.sql.DataFrame = [DocumentID: bigint, Words: array<string>]

scala> newWordsData.show(1)
+----------+--------------------+
|DocumentID|               Words|
+----------+--------------------+
|  23890098|[shlykov,, a, har...|
+----------+--------------------+
only showing top 1 row


scala> val flattened = newWordsData.withColumn("token",explode($"Words"))
flattened: org.apache.spark.sql.DataFrame = [DocumentID: bigint, Words: array<string> ... 1 more field]

scala> flattened.show
+----------+--------------------+-------------+
|DocumentID|               Words|        token|
+----------+--------------------+-------------+
|  23890098|[shlykov,, a, har...|     shlykov,|
|  23890098|[shlykov,, a, har...|            a|
|  23890098|[shlykov,, a, har...| hard-working|
|  23890098|[shlykov,, a, har...|         taxi|
|  23890098|[shlykov,, a, har...|       driver|
|  23890098|[shlykov,, a, har...|          and|
|  23890098|[shlykov,, a, har...|      lyosha,|
|  23890098|[shlykov,, a, har...|            a|
|  23890098|[shlykov,, a, har...| saxophonist,|
|  23890098|[shlykov,, a, har...|      develop|
|  23890098|[shlykov,, a, har...|            a|
|  23890098|[shlykov,, a, har...|      bizarre|
|  23890098|[shlykov,, a, har...|    love-hate|
|  23890098|[shlykov,, a, har...|relationship,|
|  23890098|[shlykov,, a, har...|          and|
|  23890098|[shlykov,, a, har...|      despite|
|  23890098|[shlykov,, a, har...|        their|
|  23890098|[shlykov,, a, har...|  prejudices,|
|  23890098|[shlykov,, a, har...|      realize|
|  23890098|[shlykov,, a, har...|         they|
+----------+--------------------+-------------+
only showing top 20 rows

Дайте мне знать, если это поможет !!