如何使用这些功能。分解以展平数据帧中的元素

Question

我编写了以下代码：

case class RawPanda(id: Long, zip: String, pt: String, happy: Boolean, attributes: Array[Double])
case class PandaPlace(name: String, pandas: Array[RawPanda])

object TestSparkDataFrame extends App{

  System.setProperty("hadoop.home.dir", "E:\\Programmation\\Libraries\\hadoop")
  val conf = new SparkConf().setAppName("TestSparkDataFrame").set("spark.driver.memory","4g").setMaster("local[*]")
  val session = SparkSession.builder().config(conf).getOrCreate()

  import session.implicits._

  def createAndPrintSchemaRawPanda(session:SparkSession):DataFrame = {
    val newPanda = RawPanda(1,"M1B 5K7", "giant", true, Array(0.1, 0.1))
    val pandaPlace = PandaPlace("torronto", Array(newPanda))
    val df =session.createDataFrame(Seq(pandaPlace))
    df
  }
  val df2 = createAndPrintSchemaRawPanda(session)
  df2.show

+--------+--------------------+
|    name|              pandas|
+--------+--------------------+
|torronto|[[1,M1B 5K7,giant...|
+--------+--------------------+


  val pandaInfo = df2.explode(df2("pandas")) {
    case Row(pandas: Seq[Row]) =>
      pandas.map{
        case (Row(
          id: Long,
          zip: String,
          pt: String,
          happy: Boolean,
          attrs: Seq[Double])) => RawPanda(id, zip, pt , happy,      attrs.toArray)
      }
  }

  pandaInfo2.show

+--------+--------------------+---+-------+-----+-----+----------+
|    name|              pandas| id|    zip|   pt|happy|attributes|
+--------+--------------------+---+-------+-----+-----+----------+
|torronto|[[1,M1B 5K7,giant...|  1|M1B 5K7|giant| true|[0.1, 0.1]|
+--------+--------------------+---+-------+-----+-----+----------+

我所使用的爆炸功能的问题已被弃用，因此我想重新计算PandaInfo2数据帧，但在警告中使用建议的方法。

请改用functions.explode

但是当我这么做的时候：

 val pandaInfo = df2.select(functions.explode(df("pandas"))

我获得了与df2相同的结果。我不知道如何继续使用平面图或函数。爆炸

如何使用平面图或函数。爆炸以获得我想要的结果？（pandaInfo中的一个）

我看过这篇文章和另一篇文章，但没有一篇对我有帮助。

Answer 1

使用explode函数调用select函数返回一个数据帧，其中数组pandas被“分解”为单独的记录；然后，如果要“展平”每个记录生成的单个“RawPanda”的结构，可以使用点分隔的“路由”选择各个列：

val pandaInfo2 = df2.select($"name", explode($"pandas") as "pandas")
  .select($"name", $"pandas",
    $"pandas.id" as "id",
    $"pandas.zip" as "zip",
    $"pandas.pt" as "pt",
    $"pandas.happy" as "happy",
    $"pandas.attributes" as "attributes"
  )

完全相同的操作的不太详细的版本是：

import org.apache.spark.sql.Encoders // going to use this to "encode" case class into schema
val pandaColumns = Encoders.product[RawPanda].schema.fields.map(_.name)

val pandaInfo3 = df2.select($"name", explode($"pandas") as "pandas")
  .select(Seq($"name", $"pandas") ++ pandaColumns.map(f => $"pandas.$f" as f): _*)