Im trying to convert a pySpark data frame to a similar one which contains struct fields. I have tried a recursive approach but cannot obtain the output structure.

In a simple way:

from pyspark.sql.types import *
from pyspark.sql import functions as f
from pyspark.sql import Row

schema_in = StructType([
    StructField("name",StringType(),True),
    StructField("age",LongType(),True),
    StructField("manager",StructType([
                          StructField("manager_name",StringType(),True),
                          StructField("manager_age",LongType(),True)]),True),
    StructField("manager_city",StringType(),True),
    StructField("manager_dept",StringType(),True)])
    
schema_out = StructType([
    StructField("name",StringType(),True),
    StructField("age",LongType(),True),
    StructField("manager",StructType([
                          StructField("manager_name",StringType(),True),
                          StructField("manager_age",LongType(),True),
                          StructField("detail",StructType([
                                               StructField("manager_city",StringType(),True),
                                               StructField("manager_dept",StringType(),True)]),True)]),True)])                                                                  

data = [
    Row("Alex", 20, Row("Bob",30),"Madrid","Supply Chain"),
    Row("Cathy", 40, Row("Doge",40),"Barcelona","Human Resources")
]

df = spark.createDataFrame(data,schema=schema_in)
df.printSchema()
df.show(20,False)

root
 |-- name: string (nullable = true)
 |-- age: long (nullable = true)
 |-- manager: struct (nullable = true)
 |    |-- manager_name: string (nullable = true)
 |    |-- manager_age: long (nullable = true)
 |-- manager_city: string (nullable = true)
 |-- manager_dept: string (nullable = true)

+-----+---+----------+------------+---------------+
|name |age|manager   |manager_city|manager_dept   |
+-----+---+----------+------------+---------------+
|Alex |20 |{Bob, 30} |Madrid      |Supply Chain   |
|Cathy|40 |{Doge, 40}|Barcelona   |Human Resources|
+-----+---+----------+------------+---------------+

df2=df.withColumn('manager', f.col('manager')
.withField('detail', f.struct(*[f.col('manager_city').alias('manager_city'), f.col('manager_dept').alias('manager_dept')])))
.drop ("manager_city")
.drop ("manager_dept")
df2.printSchema()
df2.show(20,False

root
 |-- name: string (nullable = true)
 |-- age: long (nullable = true)
 |-- manager: struct (nullable = true)
 |    |-- manager_name: string (nullable = true)
 |    |-- manager_age: long (nullable = true)
 |    |-- detail: struct (nullable = false)
 |    |    |-- manager_city: string (nullable = true)
 |    |    |-- manager_dept: string (nullable = true)

+-----+---+----------------------------------------+
|name |age|manager                                 |
+-----+---+----------------------------------------+
|Alex |20 |{Bob, 30, {Madrid, Supply Chain}}       |
|Cathy|40 |{Doge, 40, {Barcelona, Human Resources}}|
+-----+---+----------------------------------------+

What im trying to do is that it dynamically generates the data frame with the output desired schema, counting with already having input fields, so it can serve for any dataframe and output schema, providing output fields previously exists on input data frame

I got something near to the solution using this method, but I cannot figure out where is the problem (my recursivity is not any good 🙁 ):

def arrange_fields(df_in, schema_out):
    def add_new_fields(schema, prefix=""):

        columnas = []
        for field in schema.fields:
            field_name = f"{prefix}.{field.name}" if prefix else field.name
            
            if isinstance(field.dataType, StructType):
                nested_columns = add_new_fields(field.dataType, field_name)
                if prefix:
                    existing_struct = col(prefix)
                    columnas.append(struct(*nested_columns, existing_struct).alias(field.name))
                else:
                    columnas.append(struct(*nested_columns).alias(field.name))
            else:

                if field.name in df_in.columns:
                    columnas.append(col(field.name).alias(field.name))
        return columnas

    columnas = add_new_fields(schema_out)
    return df_in.select(*columnas)

df_transformed = arrange_fields(df_in, schema_out)
df_transformed.printSchema()
df_transformed.show(truncate=False)

root
 |-- name: string (nullable = true)
 |-- age: long (nullable = true)
 |-- manager: struct (nullable = false)
 |    |-- detail: struct (nullable = false)
 |    |    |-- manager_city: string (nullable = true)
 |    |    |-- manager_dept: string (nullable = true)
 |    |    |-- manager: struct (nullable = true)
 |    |    |    |-- manager_name: string (nullable = true)
 |    |    |    |-- manager_age: long (nullable = true)

+--------+---+----------------------------------+
|name    |age|manager                           |
+--------+---+----------------------------------+
|John Doe|30 |{{New York, HR, {Jane Smith, 45}}}|
+--------+---+----------------------------------+

The code updates the struct field but it adds an additional nesting level for the manager field