需要将列的值映射到另一列作为spark数据帧中的键值对

如何解决需要将列的值映射到另一列作为spark数据帧中的键值对

样本数据集：

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,56000,EEE,CBMN  
6,25000,FFF,FSS  
7,46000,GGG,LAA  
8,85000,15000,HHH,MIS      
9,90000,III,GML  
10,78000,8000,JJJ,SMA

Account1的值为Amount1，
具有值Amount2的Account2，
ACC = Amount1和Amount2之和

最终输出应为JSON格式，

{
"id":1,"AAA":10000,"ABC":20000,"ACC":30000
}  
{
"id":2,"BBB":33333,"CDE":30000,"ACC":63333
}  
{
....
....
....
....
}

解决方法

对于Scala世界

创建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)

我们还可以使用here中提到的其他scala JSON解析库。

,

您可以使用rdd.map创建字典，然后将其收集为

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"],"ACC":row["Account1"]+row["Account2"]})))
       .toDF()