这是可行的：

initial_expansion= df.withColumn("id", F.monotonically_increasing_id()).select("id","x.*");

final_df = initial_expansion\
     .join(initial_expansion.withColumn("CustomFields", F.explode("CustomFields"))\
           .select("*", "CustomFields.*")\
           .groupBy("id").pivot("Name").agg(F.first("Value")), \
        "id").drop("CustomFields")

样品输入：儿子-{'x': {'CurrentAvailableHours': 2, 'CustomFields': [{'Name': 'Country', 'Value': 'Italy'}, {'Name': 'Service', 'Value':'Dev'}]}}

输入结构：

root
 |-- x: struct (nullable = true)
 |    |-- CurrentAvailableHours: integer (nullable = true)
 |    |-- CustomFields: array (nullable = true)
 |    |    |-- element: struct (containsNull = true)
 |    |    |    |-- Name: string (nullable = true)
 |    |    |    |-- Value: string (nullable = true)

输出：

输出结构（可以删除ID）：

root
 |-- id: long (nullable = false)
 |-- CurrentAvailableHours: integer (nullable = true)
 |-- Country: string (nullable = true)
 |-- Service: string (nullable = true)

考虑下面的模型结构，类似于你的例子中的模型结构，你可以使用内联函数以sql的方式完成它：

with alpha as (
select named_struct("alpha", "abc", "beta", 2.5, "gamma", 3, "delta"
                    , array(  named_struct("a", "x", "b", "y", "c", "z")
                            , named_struct("a", "xx", "b", "yy", "c","zz"))
                   ) root
)
select root.alpha, root.beta, root.gamma, inline(root.delta) as (a, b, c) 
from alpha

结果：

实体模型结构：