Rozbijanie zagnieżdżonych struktur w ramce danych Spark

Question

Pracuję poprzez Databricks example. Schemat dla dataframe wygląda następująco:

> parquetDF.printSchema 
root 
|-- department: struct (nullable = true) 
| |-- id: string (nullable = true) 
| |-- name: string (nullable = true) 
|-- employees: array (nullable = true) 
| |-- element: struct (containsNull = true) 
| | |-- firstName: string (nullable = true) 
| | |-- lastName: string (nullable = true) 
| | |-- email: string (nullable = true) 
| | |-- salary: integer (nullable = true) 

Na przykład, pokazują jak eksplodować kolumnę pracowników do 4 dodatkowych kolumn:

val explodeDF = parquetDF.explode($"employees") { 
case Row(employee: Seq[Row]) => employee.map{ employee => 
    val firstName = employee(0).asInstanceOf[String] 
    val lastName = employee(1).asInstanceOf[String] 
    val email = employee(2).asInstanceOf[String] 
    val salary = employee(3).asInstanceOf[Int] 
    Employee(firstName, lastName, email, salary) 
} 
}.cache() 
display(explodeDF) 

Jak bym zrobić coś podobnego z kolumny dział (tj. dodać dwie dodatkowe kolumny do ramki danych o nazwie "id" i "nazwa")? Metody te nie są dokładnie takie same, a ja mogę tylko dowiedzieć się, jak utworzyć ramkę nowy danych przy użyciu:

val explodeDF = parquetDF.select("department.id","department.name") 
display(explodeDF) 

Gdy próbuję:

val explodeDF = parquetDF.explode($"department") { 
    case Row(dept: Seq[String]) => dept.map{dept => 
    val id = dept(0) 
    val name = dept(1) 
    } 
}.cache() 
display(explodeDF) 

pojawia się ostrzeżenie i błąd:

<console>:38: warning: non-variable type argument String in type pattern Seq[String] is unchecked since it is eliminated by erasure 
      case Row(dept: Seq[String]) => dept.map{dept => 
         ^
<console>:37: error: inferred type arguments [Unit] do not conform to method explode's type parameter bounds [A <: Product] 
    val explodeDF = parquetDF.explode($"department") { 
           ^

Answer 1

można użyć coś takiego:

var explodeDF = explodeDF.withColumn("id", explodeDF("department.id")) 
explodeDeptDF = explodeDeptDF.withColumn("name", explodeDeptDF("department.name")) 

które pomogły mi na te pytania i:

• Flattening Rows in Spark
• Spark 1.4.1 DataFrame explode list of JSON objects

Answer 2

To wydaje się działać (choć może nie najbardziej eleganckie rozwiązanie).

var explodeDF2 = explodeDF.withColumn("id", explodeDF("department.id")) 
explodeDF2 = explodeDF2.withColumn("name", explodeDF2("department.name"))