这并不简单，因为在分解"UPDATED_FIELDS"以创建"field"列之后，您会得到以下 Dataframe ：

+----+-------+------+-----+-------+------+-----+--------------+-----+
|s.id|s.first|s.last|s.age|d.first|d.last|d.age|UPDATED_FIELDS|field|
+----+-------+------+-----+-------+------+-----+--------------+-----+
|   1|    AAA|   BBB|   10|  AAA__|   BBB|   10|       [first]|first|
|   2|    CCC|   DDD|   20|  CCC__|   DDD|   21|  [first, age]|first|
|   2|    CCC|   DDD|   20|  CCC__|   DDD|   21|  [first, age]|  age|
+----+-------+------+-----+-------+------+-----+--------------+-----+

新的列s_value和d_value依赖于使用分解列"field"中的文字来告诉我们使用哪一列的值。

df.withColumn(
    "field", F.explode("UPDATED_FIELDS")
).withColumn(
    "s_value", F.col(f"s.{F.col('field')}")
)

...但这将导致错误，因为{F.col('field')}不能解释为文本。
更新：基于this helpful answer，实际上可以使用字符串文字作为列查找。您需要首先确保可以访问df['field']，因此需要先单独执行分解，然后使用F.coalesce(*F.when(...))作为从df['field']中的字符串获取列名中的值的方法：

df = df.withColumn(
    "field", F.explode("UPDATED_FIELDS")
)

## obtain list of all unique strings in "field"
base_cols = df.select(F.collect_set("field").alias("column")).first()["column"]

df.withColumn(
    "s_value",  F.coalesce(*[F.when(df['field'] == c, df[f"`s.{c}`"]) for c in base_cols])
).withColumn(
    "d_value",  F.coalesce(*[F.when(df['field'] == c, df[f"`d.{c}`"]) for c in base_cols])
).select(
    F.col("`s.id`").alias("id"),"field","s_value","d_value"
)

+---+-----+-------+-------+
| id|field|s_value|d_value|
+---+-----+-------+-------+
|  1|first|    AAA|  AAA__|
|  2|first|    CCC|  CCC__|
|  2|  age|     20|     21|
+---+-----+-------+-------+