问题描述
这是我先前发布的后续question。
第1步：
scala> spark.sql("select map('s1','p1','s2','p2','s3','p3') as lookup").show()
+--------------------+
| lookup|
+--------------------+
|[s1 -> p1,s2 -> ...|
+--------------------+
第2步：
scala> val df = Seq(("s1","p1"),("s2","p2"),("s3","p3")).toDF("s","p")
df: org.apache.spark.sql.DataFrame = [s: string,p: string]
scala> df.show()
+---+---+
| s| p|
+---+---+
| s1| p1|
| s2| p2|
| s3| p3|
+---+---+
第3步：
scala> val df1 = df.selectExpr("map(s,p) lookup")
df1: org.apache.spark.sql.DataFrame = [cc: map<string,string>]
scala> df1.show()
+----------+
| lookup|
+----------+
|[s1 -> p1]|
|[s2 -> p2]|
|[s3 -> p3]|
+----------+
我在步骤3中获得的预期结果是在步骤1中获得的结果。我该如何实现？
解决方法
在将键和值的两列合并为aggregated into arrays之前，它们应该为map。
import org.apache.spark.sql.functions._
df.agg(collect_list("s").as("s"),collect_list("p").as("p"))
.select(map_from_arrays('s,'p).as("lookup"))
.show(false)
输出：
+------------------------------+
|lookup |
+------------------------------+
|[s1 -> p1,s2 -> p2,s3 -> p3]|
+------------------------------+
没有collect_list调用，每行将分别转换为地图。