Scala Spark: convertir des colonnes arbitraires N en carte

Question

je la structure de données suivantes représentant ids de film (première colonne) et notes pour différents utilisateurs pour ce film dans le reste des colonnes - quelque chose comme ça:

+-------+----+----+----+----+----+----+----+----+----+----+----+----+----+----+----+ 
|movieId| 1| 2| 3| 4| 5| 6| 7| 8| 9| 10| 11| 12| 13| 14| 15| 
+-------+----+----+----+----+----+----+----+----+----+----+----+----+----+----+----+ 
| 1580|null|null| 3.5| 5.0|null|null|null|null|null|null|null|null|null|null|null| 
| 3175|null|null|null|null|null|null|null|null|null|null|null|null|null| 5.0|null| 
| 3794|null|null|null|null|null|null|null|null|null|null|null| 3.0|null|null|null| 
| 2659|null|null|null| 3.0|null|null|null|null|null|null|null|null|null|null|null| 

Je veux convertir cette trame de données à un DataSet de

final case class MovieRatings(movie_id: Long, ratings: Map[Long, Double])

Alors que ce serait quelque chose comme

[1580, [1 -> null, 2 -> null, 3 -> 3.5, 4 -> 5.0, 5 -> null, 6 -> null, 7 -> null,...]] 

Etc.

Comment cela peut être fait?

La chose ici est que le nombre d'utilisateurs est arbitraire. Et je veux les compresser en une seule colonne en laissant la première colonne intacte.

Answer 1

D'abord, vous devez Tranform votre dataframe en un seul avec un schéma correspondant à votre classe de cas, vous pouvez utiliser .as[MovieRatings] pour convertir dataframe en Dataset[MovieRatings]:

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

// define a new MapType column using `functions.map`, passing a flattened-list of 
// column name (as a Long column) and column value 
val mapColumn: Column = map(df.columns.tail.flatMap(name => Seq(lit(name.toLong), $"$name")): _*) 

// select movie id and map column with names matching the case class, and convert to Dataset: 
df.select($"movieId" as "movie_id", mapColumn as "ratings") 
    .as[MovieRatings] 
    .show(false) 

Answer 2

Vous pouvez utiliser les spark.sql.functions .map pour créer une carte à partir de colonnes arbitraires. Il attend une séquence alternant entre les clés et les valeurs qui peuvent être des types de colonnes ou des chaînes. Voici un exemple:

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

case class Input(movieId: Int, a: Option[Double], b: Option[Double], c: Option[Double]) 

val data = Input(1, None, Option(3.5), Option(1.4)) :: 
     Input(2, Option(4.2), Option(1.34), None) :: 
     Input(3, Option(1.11), None, Option(3.32)) :: Nil 

val df = sc.parallelize(data).toDF 

// Exclude the PK column from the map 
val mapKeys = df.columns.filterNot(_ == "movieId") 

// Build the sequence of key, value, key, value, .. 
val pairs = mapKeys.map(k => Seq(lit(k), col(k))).flatten 

val mapped = df.select($"movieId", functions.map(pairs:_*) as "map") 
mapped.show(false) 

produit cette sortie:

+-------+------------------------------------+ 
|movieId|map         | 
+-------+------------------------------------+ 
|1  |Map(a -> null, b -> 3.5, c -> 1.4) | 
|2  |Map(a -> 4.2, b -> 1.34, c -> null) | 
|3  |Map(a -> 1.11, b -> null, c -> 3.32)| 
+-------+------------------------------------+