Отдельное количество слов в строке

Question

Это немного отличается от обычной программы подсчета слов. Я пытаюсь получить различное количество слов в строке.

Ввод :

Line number one has six words
Line number two has two words

Ожидаемый вывод :

line1 => (Line,1),(number,1),(one,1),(has,1),(six,1),(words,1)
line2 => (Line,1),(number,1),(two,2),(has,1),(words,1)

Может кто-нибудь, пожалуйста, направьте меня.

Answer 1

Используя Dataframe во встроенных функциях explode,split,collect_set,groupBy.

//input data
val df=Seq("Line number one has six words","Line number two has has two words").toDF("input")

scala> :paste
// Entering paste mode (ctrl-D to finish)

df.withColumn("words",explode(split($"input","\\s+"))) //split by space and explode
.groupBy("input","words") //group by on both columns
.count()
.withColumn("line_word_count",struct($"words",$"count")) //create struct
.groupBy("input") //grouping by input data column
.agg(collect_set("line_word_count").alias("line_word_count"))
.show(false)

Result:

+---------------------------------+------------------------------------------------------------------+
|input                            |line_word_count                                                   |
+---------------------------------+------------------------------------------------------------------+
|Line number one has six words    |[[one, 1], [has, 1], [six, 1], [number, 1], [words, 1], [Line, 1]]|
|Line number two has has two words|[[has, 2], [two, 2], [words, 1], [number, 1], [Line, 1]]          |
+---------------------------------+------------------------------------------------------------------+

---

Если вы ожидаете номера строк, то используйте concat,monotonically_increasing_id функции:

df.withColumn("line",concat(lit("line"),monotonically_increasing_id()+1))
.withColumn("words",explode(split($"input","\\s+"))) 
.groupBy("input","words","line") 
.count() 
.withColumn("line_word_count",struct($"words",$"count")) 
.groupBy("line") 
.agg(collect_set("line_word_count").alias("line_word_count")) 
.show(false)

Result:

+-----+------------------------------------------------------------------+
|line |line_word_count                                                   |
+-----+------------------------------------------------------------------+
|line1|[[one, 1], [has, 1], [six, 1], [words, 1], [number, 1], [Line, 1]]|
|line2|[[has, 2], [two, 2], [number, 1], [words, 1], [Line, 1]]          |
+-----+------------------------------------------------------------------+

Примечание в случае больших наборов данных чтобы сделать это последовательно, нам нужно сделать .repartition (1) .

Answer 2

Вот еще один способ использования RDD API:

val rdd = df.withColumn("output", split($"input", " ")).rdd.map(l => (
                l.getAs[String](0), 
                l.getAs[Seq[String]](1).groupBy(identity).mapValues(_.size).map(identity))
          )

val dfCount = spark.createDataFrame(rdd).toDF("input", "output")

Не большой поклонник использования UDF, но это также можно сделать так:

import org.apache.spark.sql.functions.udf

val mapCount: Seq[String] => Map[String, Integer] = _.groupBy(identity).mapValues(_.size)
val countWordsUdf  = udf(mapCount)

df.withColumn("output", countWordsUdf(split($"input", " "))).show(false)

Дает:

+---------------------------------+------------------------------------------------------------------+
|input                            |output                                                            |
+---------------------------------+------------------------------------------------------------------+
|Line number one has six words    |[number -> 1, Line -> 1, has -> 1, six -> 1, words -> 1, one -> 1]|
|Line number two has has two words|[number -> 1, two -> 2, Line -> 1, has -> 2, words -> 1]          |
+---------------------------------+------------------------------------------------------------------+