• 使用split（col（“id_list”），“，”）在逗号上拆分df_B中的“id_list”列，然后使用explode（）为拆分后的每个元素生成一行。分解的元素存储在称为“id”的新列中。
• 使用df_a.id == df_b. id左连接df_a和“id”列上修改后的df_b。结果存储在joined_df中。
• 将joined_df按“id_list”列分组，并使用collect_list（“names”）聚合“names”列。生成的聚合使用别名（“users”）存储在名为“users”的新列中。

代码：

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, split, explode, collect_list, concat_ws
spark = SparkSession.builder.getOrCreate()
data_a = [("abc-2", "John"), ("bbb-1", "Peter"), ("def-y", "Kim"), ("xyz-0", "Brad")]
df_a = spark.createDataFrame(data_a, ["id", "names"])
data_b = [("abc-2,bbb-1",), ("abc-2",), ("abc-2,xyz-0",)]
df_b = spark.createDataFrame(data_b, ["id_list"])
 
df_b = df_b.withColumn("id", explode(split(col("id_list"), ",")))
joined_df = df_b.join(df_a, df_a.id == df_b.id, "left")
final_df = joined_df.groupby("id_list").agg(concat_ws(",", collect_list("names")).alias("users"))
final_df = final_df.withColumn("users", final_df["users"].cast("string"))
final_df.display(truncate=False)