La sortie de Spark DataFrame remplace les noms de colonnes par "0", "1" lorsque les données arrays_zip sont imbriquées.

Question

J'ai utilisé les fonctions spark sql arrays_zip combinées avec flatten pour transformer les données d'un tableau de struct d'un tableau interne de même longueur en un tableau de struct. printSchema montre exactement ce que je veux. Cependant, la sortie df a perdu les noms de colonnes originaux et les a remplacés par des noms de colonnes génériques "0", "1", "2" etc., peu importe le format Parquet ou Avro. J'aimerais que les noms de colonnes originaux soient affichés.

Ne pas révéler l'activité de mon entreprise. Les exemples suivants sont similaires mais beaucoup plus simples.

scala> c2.printSchema
root
 |-- cal: array (nullable = true)
 |    |-- element: struct (containsNull = true)
 |    |    |-- month: array (nullable = true)
 |    |    |    |-- element: string (containsNull = true)
 |    |    |-- num: array (nullable = true)
 |    |    |    |-- element: long (containsNull = true)
scala> c2.show(false)
+----------------------------------------------+
|cal                                           |
+----------------------------------------------+
|[[[Jan, Feb, Mar], [1, 2, 3]], [[April], [4]]]|
+----------------------------------------------+

J'aime me transformer pour

scala> newC2.show(false)
+------------------------------------------+
|cal                                       |
+------------------------------------------+
|[[Jan, 1], [Feb, 2], [Mar, 3], [April, 4]]|
+------------------------------------------+
with
scala> newC2.printSchema
root
 |-- cal: array (nullable = true)
 |    |-- element: struct (containsNull = false)
 |    |    |-- month: string (nullable = true)
 |    |    |-- num: long (nullable = true)

Je sais que arrays_zip ne fonctionne bien que sur les tableaux de premier niveau. Par conséquent, je les aplatis au niveau supérieur. Les codes suivants fonctionnent dans cet exemple

val newC2 = c2.withColumn("month", flatten(col("cal.month"))).withColumn("num", flatten(col("cal.num"))).withColumn("cal", arrays_zip(col("month"), col("num"))).drop("month", "num")

Il génère exactement les données et le schéma que je veux. Cependant, il produit toutes les colonnes de manière générique en utilisant "0", "1", "2", etc.

newC2.write.option("header", false).parquet("c2_parquet")

J'ai essayé un autre exemple qui a les données originales du tableau month et du tableau num au niveau supérieur. Je peux arrays_zip sans flatten et obtenir le même schéma et les mêmes données. Cependant, le nom du champ original est correctement affiché dans ce cas.

J'ai essayé d'ajouter un alias pour aplatir les données. Cela ne fonctionne pas. J'ai même essayé de manipuler des colonnes comme (supposer que le champ stocker le résultat de arrays_zip est 'zipped').

val columns: Array[Column] = inner.fields.map(_.name).map{x => col("zipped").getField(x).alias(x)}
    val newB3 = newB2.withColumn("b", array(struct(columns:_*))).drop("zipped")

Il finit par générer le schéma original ("month", tableau de string et "num", tableau de long).

Pour dupliquer le problème, vous pouvez utiliser l'entrée json

"cal":[{"month":["Jan","Feb","Mar"],"num":[1,2,3]},{"month":["April"],"num":[4]}]}

le json suivant est pour le niveau supérieur de arrays_zip

{"month":["Jan","Feb","Mar"],"num":[1,2,3]}

Comment Spark décide-t-il en interne des noms de champs à utiliser ? Comment puis-je faire en sorte que cela fonctionne ? Veuillez me conseiller.

Answer 1

Depuis Spark 2.4, la transformation du schéma peut être réalisée à l'aide de fonctions d'ordre supérieur. En Scala, la requête peut ressembler à ceci :

import org.apache.spark.sql.functions.{expr, flatten}

val result = df
.withColumn("cal", flatten(expr("TRANSFORM(cal, x -> zip_with(x.month, x.num, (month, num) -> (month,num)))")))

Après avoir appliqué sur votre échantillon de données, j'ai obtenu ce schéma :

result.printSchema()
root
 |-- cal: array (nullable = true)
 |    |-- element: struct (containsNull = false)
 |    |    |-- month: string (nullable = true)
 |    |    |-- num: long (nullable = true)