Nombre de valeurs de liste dans spark - dataframe

Question

Dans Cassandra, j'ai une colonne de type liste. Je suis novice en spark et scala, et je ne sais pas du tout par où commencer. Dans spark je veux obtenir le compte de chaque valeur, est-ce possible de le faire. Voici le dataframe

+--------------------+------------+
|                  id|        data|
+--------------------+------------+
|53e5c3b0-8c83-11e...|      [b, c]|
|508c1160-8c83-11e...|      [a, b]|
|4d16c0c0-8c83-11e...|   [a, b, c]|
|5774dde0-8c83-11e...|[a, b, c, d]|
+--------------------+------------+

Je souhaite obtenir le résultat suivant

+--------------------+------------+
|   value            |      count |
+--------------------+------------+
|a                   |      3     |
|b                   |      4     |
|c                   |      3     |
|d                   |      1     |
+--------------------+------------+

Version de l'étincelle : 1.4

Answer 1

Nous y voilà :

scala> val rdd = sc.parallelize(
  Seq(
    ("53e5c3b0-8c83-11e", Array("b", "c")),
    ("53e5c3b0-8c83-11e1", Array("a", "b")),
    ("53e5c3b0-8c83-11e2", Array("a", "b", "c")),
    ("53e5c3b0-8c83-11e3", Array("a", "b", "c", "d"))))
// rdd: org.apache.spark.rdd.RDD[(String, Array[String])] = ParallelCollectionRDD[22] at parallelize at <console>:27

scala> rdd.flatMap(_._2).map((_, 1)).reduceByKey(_ + _)
// res11: org.apache.spark.rdd.RDD[(String, Int)] = ShuffledRDD[21] at reduceByKey at <console>:30

scala> rdd.flatMap(_._2).map((_,1)).reduceByKey(_ + _).collect
// res16: Array[(String, Int)] = Array((a,3), (b,4), (c,3), (d,1))

C'est également très facile avec l'API DataFrame :

scala> val df = rdd.toDF("id", "data")
// res12: org.apache.spark.sql.DataFrame = ["id": string, "data": array<string>]

scala> df.select(explode($"data").as("value")).groupBy("value").count.show
// +-----+-----+
// |value|count|
// +-----+-----+
// |    d|    1|
// |    c|    3|
// |    b|    4|
// |    a|    3|
// +-----+-----+

Answer 2

Vous avez besoin de quelque chose comme ceci (de Exemples Apache Spark ):

val textFile = sc.textFile("hdfs://...")
val counts = textFile
             .flatMap(line => line.split(" "))
             .map(word => (word, 1))
             .reduceByKey(_ + _)

En supposant que vous avez déjà des paires, .reduceByKey(_ + _) retournera ce dont vous avez besoin.

Vous pouvez également essayer dans spark shell quelque chose comme ceci :

sc.parallelize(Array[Integer](1,1,1,2,2),3).map(x=>(x,1)).reduceByKey(_+_).foreach(println)