Сведение вложенного фрейма данных

Поскольку вы работаете с массивами, вы не можете просто выбрать подстолбцы: сначала вам нужно будет использовать explode.

Этот блок кода предназначен только для воссоздания ваших данных (вам не нужно этого делать):

import datetime
from pyspark.sql.types import *

schema = StructType([
    StructField("date_time", TimestampType(), True),
    StructField("filename", StringType(), True),
    StructField("label", StringType(), True),
    StructField("description", StringType(), True),
    StructField("feature_set",
        ArrayType(
            StructType([
                StructField("direction", StringType(), True),
                StructField("tStart", DoubleType(), True),
                StructField("tEnd", DoubleType(), True),
                StructField("features",
                    ArrayType(
                        StructType([
                            StructField("field1", StringType(), True),
                            StructField("field2", StringType(), True),
                            StructField("field3", StringType(), True),
                            StructField("field4", StringType(), True),
                        ])
                    ), True),
            ])
        )
    )
])

data = [ ( datetime.datetime(2022, 8, 24, 7, 51, 54), 'filename1', 'label1', 'description of file 1', [ ( 'east', 78.23018987, 79.23010199, [ ( 'fld_val11', 'fld_val12', 'fld_Val13', 'fld_Val14') ]), ( 'west', 78.23018987, 79.23010199, [ ( 'fld_val21', 'fld_val22', 'fld_val23', 'fld_val24') ]), ( 'south', 78.23018987, 79.23010199, [ ( 'fld_val31', 'fld_val32', 'fld_val33', 'fld_val34') ]) ]) ]

df = spark.createDataFrame(data, schema)

Теперь, когда у нас есть данные, нам нужно использовать комбинацию select и explode, чтобы получить нужную схему:

from pyspark.sql import functions as F

flat_df = df \
  .withColumn("feature_set", F.explode("feature_set")) \
  .select("date_time", "filename", "label", "description", "feature_set.*") \
  .withColumn("features", F.explode("features")) \
  .select("date_time", "filename", "label", "description", "direction", "tStart", "tEnd", "features.*")

Итак, как вы видите, мы explode создаем каждый столбец массива и затем разворачиваем их. Таким образом, вы говорите Spark создать строку в результирующем фрейме данных для каждого элемента в имеющихся у вас массивах.

Выходная схема не будет точно такой же, как то, что вы запрашиваете (например, массива feature_set больше нет), но это делает более компактную версию без дублирования данных. Результирующая схема и фрейм данных выглядят следующим образом:

>>> flat_df.printSchema()
root
 |-- date_time: timestamp (nullable = true)
 |-- filename: string (nullable = true)
 |-- label: string (nullable = true)
 |-- description: string (nullable = true)
 |-- direction: string (nullable = true)
 |-- tStart: double (nullable = true)
 |-- tEnd: double (nullable = true)
 |-- field1: string (nullable = true)
 |-- field2: string (nullable = true)
 |-- field3: string (nullable = true)
 |-- field4: string (nullable = true)

>>> flat_df.show()
+-------------------+---------+------+--------------------+---------+-----------+-----------+---------+---------+---------+---------+
|          date_time| filename| label|         description|direction|     tStart|       tEnd|   field1|   field2|   field3|   field4|
+-------------------+---------+------+--------------------+---------+-----------+-----------+---------+---------+---------+---------+
|2022-08-24 07:51:54|filename1|label1|description of fi...|     east|78.23018987|79.23010199|fld_val11|fld_val12|fld_Val13|fld_Val14|
|2022-08-24 07:51:54|filename1|label1|description of fi...|     west|78.23018987|79.23010199|fld_val21|fld_val22|fld_val23|fld_val24|
|2022-08-24 07:51:54|filename1|label1|description of fi...|    south|78.23018987|79.23010199|fld_val31|fld_val32|fld_val33|fld_val34|
+-------------------+---------+------+--------------------+---------+-----------+-----------+---------+---------+---------+---------+