试试这个，

import org.apache.spark.sql.functions._

 df
  .withColumn("attributes_splitted",  split(col("person_attributes"), "__")) // Split by delimiter `__`
  .withColumn("exploded", explode(col("attributes_splitted"))) // explode the splitted column
  .withColumn("temp", split(col("exploded"), "=")) // again split based on delimiter `=`
  .withColumn("attribute_key", col("temp").getItem(0))
  .withColumn("attribute_value", col("temp").getItem(1))
  .drop("attributes_splitted", "exploded", "temp", "person_attributes")
  .show(false)

试试这个 Spark2.3 :

from pyspark.sql import functions as F

df.withColumn("arr", F.split("person_attributes",'\=|__'))\
  .withColumn("map", F.create_map(F.lit('department'),F.col("arr")[1]\
                                  ,F.lit('title'),F.col("arr")[3]))\
  .select("person_id", F.explode("map").alias("attribute_key","attribute_value"))\
  .show(truncate=False)

# +---------+-------------+-----------------+

# |person_id|attribute_key|attribute_value  |

# +---------+-------------+-----------------+

# |id_1     |department   |Sales            |

# |id_1     |title        |Sales_executive  |

# |id_2     |department   |Engineering      |

# |id_2     |title        |Software Engineer|

# +---------+-------------+-----------------+

试试这个 Spark2.4+ ```
from pyspark.sql import functions as F

df.withColumn("arr", F.split("person_attributes",'=|__'))
.withColumn("map", F.map_from_arrays(F.expr("""filter(arr,(x,i)->i%2=0)""")
,F.expr("""filter(arr,(x,i)->i%2!=0)""")))
.select("person_id", F.explode("map").alias("attribute_key","attribute_value")).filter("""attribute_key!='level'""")
.show(truncate=False)

+---------+-------------+-----------------+

|person_id|attribute_key|attribute_value |

+---------+-------------+-----------------+

|id_1 |department |Sales |

|id_1 |title |Sales_executive |

|id_2 |department |Engineering |

|id_2 |title |Software Engineer|

+---------+-------------+-----------------+