как трансформировать List(MovieCritics(name: String, movieRatings: List[MovieRatings]))
Это так же просто, как использовать toDS на List. Это доступно, только если у вас есть имплициты в области действия, которая (опять же) так же проста, как следующее:
val sparkSession = SparkSession.builder.getOrCreate()
import sparkSession.implicits._
Если вы работаете с scala.collection.immutable.Iterable[MovieCritics] или аналогичной структурой данных коллекции, вы должны «отобразить» ее, используя toSeq или toArray перед toDS, чтобы «убежать» из Iterable. Неявные значения недоступны для Iterables.
Учитывая, что список critics, вам нужно будет сделать следующее:
critics.toDS
Теперь я хочу преобразовать этот список в искровой фрейм данных, чтобы отображать данные в более удобном для пользователя виде.
Это самая интересная часть вашего вопроса (мне потребовалось пару часов, чтобы наконец понять и написать решение). Я был бы признателен за комментарии, чтобы сделать его красивее.
case class MovieRatings(movieName: String, rating: Double)
case class MovieCritics(name: String, movieRatings: Seq[MovieRatings])
val movies_critics = Seq(
MovieCritics("Manuel", Seq(MovieRatings("Logan", 1.5), MovieRatings("Zoolander", 3), MovieRatings("John Wick", 2.5))),
MovieCritics("John", Seq(MovieRatings("Logan", 2), MovieRatings("Zoolander", 3.5), MovieRatings("John Wick", 3))))
С набором входных данных приходит решение.
val ratings = movies_critics.toDF
scala> ratings.show(false)
+------+-----------------------------------------------+
|name |movieRatings |
+------+-----------------------------------------------+
|Manuel|[[Logan,1.5], [Zoolander,3.0], [John Wick,2.5]]|
|John |[[Logan,2.0], [Zoolander,3.5], [John Wick,3.0]]|
+------+-----------------------------------------------+
val ratingsCount = ratings.
withColumn("size", size($"movieRatings")).
select(max("size")).
as[Int].
head
val names_ratings = (0 until ratingsCount).
foldLeft(ratings) { case (ds, counter) => ds.
withColumn(s"name_$counter", $"movieRatings"(counter)("movieName")).
withColumn(s"rating_$counter", $"movieRatings"(counter)("rating")) }
val movieColumns = names_ratings.
columns.
drop(1).
filter(name => name.startsWith("name")).
map(col)
val movieNames = names_ratings.select(movieColumns: _*).head.toSeq.map(_.toString)
val ratingNames = movieNames.indices.map(idx => s"rating_$idx")
val cols = movieNames.zip(ratingNames).map { case (movie, rn) =>
col(rn) as movie
}
val solution = names_ratings.select(($"name" +: cols): _*)
scala> solution.show
+------+-----+---------+---------+
| name|Logan|Zoolander|John Wick|
+------+-----+---------+---------+
|Manuel| 1.5| 3.0| 2.5|
| John| 2.0| 3.5| 3.0|
+------+-----+---------+---------+
27.06.2017
