我在下面有一个spark数据框：

df.show()
+------+--------------------+--------------------+------+--------+------------------+
|    id|            A_amount|            B_amount|  A_pi|    B_pi|mismatchingColumns|
+------+--------------------+--------------------+------+--------+------------------+
|     1|5.000000000000000000|5.000003000000000000|   PI1|     PI1|          [amount]|
|     2|5.000000000000000000|5.000000000000000000|   PI2|     PI3|              [pi]|
|     3|1.000000000000000000|1.000001000000000000|   PI4|     PI5|      [amount, pi]|
+------+--------------------+--------------------+------+--------+------------------+

我想做的是创建一个新的列，使用中的每个列名 mismatchingColumns 动态地访问dataframe中带有“a”和“b”前缀的其他列，并将它们放入Map中。

+------------------+---------------------------------------------------------------------+
|mismatchingColumns|                                                       New Map Column|
+------------------+---------------------------------------------------------------------+
|          [amount]|               [amount -> 5.000000000000000000, 5.000003000000000000]|
|              [pi]|                                                    [pi -> PI2, PI3 ]|
|      [amount, pi]|[amount -> 1.000000000000000000, 1.000001000000000000, pi -> PI4, PI5]
+------------------+---------------------------------------------------------------------+

我考虑过使用 transform 上 mismatchingColumns 列，但由于lambda函数的类型为 Column => Column ，而 col 是 String => Column .
我正在运行spark版本 3.0.1 scala版本scalaversion 2.12.12 .
任何帮助都非常感谢-谢谢。