Scala Funken: beliebig N Spalten in

Question

Map umwandeln habe ich die Struktur folgende Daten, welche Film-IDs (erste Spalte) und Ratings für verschiedene Benutzer für diesen Film in den Rest der Spalten - so ähnlich:

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

ich möchte diese Datenrahmen zu einem Datensatz von

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

So konvertieren, dass es so etwas wie

wäre

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

Etc.

Wie dies geschehen kann?

Die Sache hier ist, dass die Anzahl der Benutzer ist beliebig. Und ich möchte diese in eine einzelne Spalte zippen und die erste Spalte unberührt lassen.

Answer 1

Zuerst müssen Sie Ihre Datenrahmen in eine mit einem Schema für Ihre Fallklasse tranform, dann können Sie .as[MovieRatings] Datenrahmen zu konvertieren verwenden, um in ein 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

Sie die spark.sql.functions verwenden können .map, um eine Karte aus beliebigen Spalten zu erstellen. Es erwartet eine Sequenz, die zwischen Schlüsseln und Werten wechselt, die Spaltentypen oder Zeichenketten sein können. Hier ein Beispiel:

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) 

diese Ausgabe erzeugt:

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