在 Scala 中将 Spark 的 DataFrame 列转换为 List[String]

Question

我正在处理 Movie Lens 数据集。在 csv 个文件中，数据结构为：

movieId movieTitle genres

和 genres 也是 | 分隔值的列表，该字段可以为空。

我正在尝试获取所有 genres 的唯一列表，以便我可以按如下方式重新排列数据：

movieId movieTitle genre1 genre2 ... genreN

和一行，其中 genre 和 genre1 | genre2 看起来像：

1Title1110...0

到目前为止，我已经能够使用以下代码读取 csv 文件：

val conf         = new SparkConf().setAppName(App.name).setMaster(App.sparkMaster)
val context      = new SparkContext(conf)
val sparkSession = SparkSession.builder()
                   .appName(App.name)
                   .config("header", "true")
                   .config(conf = conf)
                   .getOrCreate()

val movieFrame: DataFrame = sparkSession.read.csv(moviesPath)

如果我尝试这样的事情：

movieFrame.rdd.map(row ⇒ row(2).asInstanceOf[String]).collect()

然后我得到以下异常：

java.lang.ClassNotFoundException: com.github.babbupandey.ReadData$$anonfun

然后，此外，我尝试使用以下代码显式提供架构：

val moviesSchema: StructType = StructType(Array(StructField("movieId", StringType, nullable = true),
                                                        StructField("title", StringType, nullable = true),
                                                        StructField("genres", StringType, nullable = true)))

并尝试过：

val movieFrame: DataFrame = sparkSession.read.schema(moviesSchema).csv(moviesPath)

然后我遇到了同样的异常。

有什么方法可以将 genres 的集合设置为 List 或 Set 以便我可以进一步将数据转换为所需的格式？任何帮助将不胜感激。

Answer 1

我是这样得到这组流派的：

val genreList: Array[String] = for (row <- movieFrame.select("genres").collect) yield row.getString(0)
val genres: Array[String] =  for {
        g ← genreList
        genres ← g.split("\|")
    } yield genres
val genreSet : Set[String] = genres.toSet

Answer 2

这有效地给出了一个 Array[Array[String]]

    val genreLst = movieFrame.select("genres").rdd.map(r =>     r(0).asInstanceOf[String].split("\|").map(_.toString).distinct).collect()

获取数组[String]

    val genres = genreLst.flatten

或

    val genreLst = movieFrame.select("genres").rdd.map(r => r(0).asInstanceOf[String].split("\|").map(_.toString).distinct).collect().flatten

在 Scala 中将 Spark 的 DataFrame 列转换为 List[String]

Converting a Spark's DataFrame column to List[String] in Scala

csv

scala

dataframe

apache-spark

spark-dataframe