您需要在分组后使用相关Map功能创建Map：

val df2 = df.groupBy("User", "country")
  .count()
  .groupBy("User")
  .agg(map(
      col("User"), 
      map_from_entries(collect_list(struct(col("country"), col("count"))))
      ).as("result")
  )
  .select("result")

df2.show(false)
+---------------------------------+
|result                           |
+---------------------------------+
|[Tom -> [china -> 3, japan -> 1]]|
|[Lin -> [spain -> 1, japan -> 2]]|
|[Ron -> [italy -> 1]]            |
+---------------------------------+

如果要将它们全部放在一行中，可以再进行一次聚合：

val df2 = df.groupBy("User", "country")
  .count()
  .groupBy("user")
  .agg(map_from_entries(collect_list(struct(col("country"), col("count")))).as("result"))
  .agg(map_from_entries(collect_list(struct(col("user"), col("result")))).as("result_all"))

df2.show(false)
+---------------------------------------------------------------------------------------+
|result_all                                                                             |
+---------------------------------------------------------------------------------------+
|[Tom -> [china -> 3, japan -> 1], Lin -> [spain -> 1, japan -> 2], Ron -> [italy -> 1]]|
+---------------------------------------------------------------------------------------+