星火斯卡拉：轉換任意N列到地圖

Question

我代表電影IDS（第一列）和評級爲不同的用戶對於電影中的其餘列如下數據結構 - 這樣的事情：

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

我想這個數據幀因此，它會像

轉換成的

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

一個DataSet

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

等等

這可怎麼辦呢？

這裏的事情是用戶數量是任意的。我想將這些壓縮到一個列中，並保持第一列不變。

Answer 1

首先，你必須要變換分析你的數據幀到一個與模式匹配您的案例類，那麼你可以使用.as[MovieRatings]到數據幀轉換爲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

可以使用spark.sql.functions .map從任意列創建地圖。它期望在鍵和值之間交替的序列可以是列類型或字符串。下面是一個例子：

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) 

產生這樣的輸出：

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