你可以用 rdd.map 创建一个dict，然后你可以收集它作为

df.rdd.map(lambda row: {"id" :row["id"], 
           row["Account1"]:row["Amount1"],
           row["Account2"]:row["Amount2"],
          "ACC":row["Account1"]+row["Account2"]})
       .toDF()

对于json，可以使用 json.dumps() 但是它会使每一行都成为字符串，所以您可以使用这个hack来传递它 json.loads 如下所示

import json
df.rdd.map(lambda row: json.loads(json.dumps({"id" :row["id"], 
           row["Account1"]:row["Amount1"],
           row["Account2"]:row["Amount2"],
          "ACC":row["Account1"]+row["Account2"]})))
       .toDF()

对于scala world
创建 case class Input(id:String,Amount1:String,Amount2:String,Account1:String,Account2:String) ```
val df = spark.read.option("header",true).csv(inputFile).as[Input]
df.show
+---+-------+-------+--------+--------+
| id|Amount1|Amount2|Account1|Account2|
+---+-------+-------+--------+--------+
| 1| 10000| 20000| AAA| ABC|
| 2| 33333| 30000| BBB| CDE|
| 3| 50000| 45000| CCC| BAA|
| 4| 60000| 65600| DDD| DSP|
| 5| 45000| 56000| EEE| CBMN|
| 6| 20000| 25000| FFF| FSS|
| 7| 46000| 25000| GGG| LAA|
| 8| 85000| 15000| HHH| MIS |
| 9| 90000| 10000| III| GML|
| 10| 78000| 8000| JJJ| SMA|
+---+-------+-------+--------+--------+

像这样一个简单的转换会给你一个结果。

df.map(x => {
s"""{
"id": ${x.id.trim},
"${x.Account1}" : ${x.Amount1.trim},
"${x.Account2}" : ${x.Amount2.trim},
"ACC" : ${x.Amount1.trim.toInt+x.Amount2.trim.toInt}
}""".stripMargin
}).show(false)

我们还可以使用这里提到的其他scala json解析库。