[FIXED] Explodieren Sie das JSON-Array in Zeilen

Ausgabe

Ich habe einen Datenrahmen mit 2 Spalten “ID” und “input_array” (Werte sind JSON-Arrays).

ID   input_array
1    [ {“A”:300, “B”:400}, { “A”:500,”B”: 600} ]
2    [ {“A”: 800, “B”: 900} ]

Ausgabe, die ich brauche:

ID A      B
1  300    400
1  500    600
2  800    900

Ich habe versucht from_json, explodeFunktionen. Bei Array-Spalten tritt jedoch ein Datentyp-Nichtübereinstimmungsfehler auf.

Echtes

Datenbild Im Bild ist der 1. Datenrahmen der Eingabedatenrahmen, den ich lesen und in den 2. Datenrahmen konvertieren muss. 3 Eingabezeilen müssen in 5 Ausgabezeilen konvertiert werden.

Lösung

Ich habe 2 Interpretationen dessen, welche Eingabedatentypen (Spalte “input_array”) Sie haben.

• Wenn es eine Schnur ist…

  df = spark.createDataFrame(
      [(1, '[ {"A":300, "B":400}, { "A":500,"B": 600} ]'),
       (2, '[ {"A": 800, "B": 900} ]')],
      ['ID', 'input_array'])
  df.printSchema()
  # root
  #  |-- ID: long (nullable = true)
  #  |-- input_array: string (nullable = true)
  

  … können Sie verwenden from_json, um die Spark-Struktur aus der JSON-Zeichenfolge zu extrahieren und dann inlinedas resultierende Array von Strukturen in Spalten aufzulösen.

  df = df.selectExpr(
      "ID",
      "inline(from_json(input_array, 'array<struct<A:long,B:long>>'))"
  )
  df.show()
  # +---+---+---+
  # | ID|  A|  B|
  # +---+---+---+
  # |  1|300|400|
  # |  1|500|600|
  # |  2|800|900|
  # +---+---+---+
  

• Wenn es ein Array von Strings ist …

  df = spark.createDataFrame(
      [(1, [ '{"A":300, "B":400}', '{ "A":500,"B": 600}' ]),
       (2, [ '{"A": 800, "B": 900}' ])],
      ['ID', 'input_array'])
  df.printSchema()
  # root
  #  |-- ID: long (nullable = true)
  #  |-- input_array: array (nullable = true)
  #  |    |-- element: string (containsNull = true)
  

  … können Sie zuerst verwenden explode, um das Element jedes Arrays in Zeilen zu verschieben, was zu einer Spalte vom Typ Zeichenfolge führt, dann verwenden from_json, um Spark-Datentypen aus den Zeichenfolgen zu erstellen und schließlich *die Strukturen in Spalten zu erweitern.

  from pyspark.sql import functions as F
  
  df = df.withColumn('input_array', F.explode('input_array'))
  df = df.withColumn('input_array', F.from_json('input_array', 'struct<A:long,B:long>'))
  df = df.select('ID', 'input_array.*')
  
  df.show()
  # +---+---+---+
  # | ID|  A|  B|
  # +---+---+---+
  # |  1|300|400|
  # |  1|500|600|
  # |  2|800|900|
  # +---+---+---+
  

Beantwortet von – ZygD

Antwort geprüft von – Mildred Charles (FixError Admin)