你可以用 to_json 按分组后 User Id 创建Map列：

from pyspark.sql import functions as F

df1 = df.groupBy("User Id").agg(*[
    F.to_json(
        F.map_from_entries(F.collect_list(F.struct(F.col("Product"), F.col(c))))
    ).alias(c)
    for c in df.columns[2:]
])

df1.show()

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

# |User Id|         Amount 1|         Amount 2|         Amount 3|

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

# |      1|{"A":100,"B":200}|{"A":200,"B":300}|{"A":300,"B":400}|

# |      2|        {"A":500}|        {"A":600}|        {"A":700}|

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