SparkSql系列(8/25) 行列旋转

共计 2087 个字符，预计需要花费 6 分钟才能阅读完成。

pivot 函数的功能是实现行列旋转，在 sql 里面也经常会遇到行列旋转，相应的操作大概就是对于行你可能需要解包拆成多列，多列变成一行

构建 DataFrame

val data = Seq(("Banana",1000,"USA"), ("Carrots",1500,"USA"), ("Beans",1600,"USA"),
      ("Orange",2000,"USA"),("Orange",2000,"USA"),("Banana",400,"China"),
      ("Carrots",1200,"China"),("Beans",1500,"China"),("Orange",4000,"China"),
      ("Banana",2000,"Canada"),("Carrots",2000,"Canada"),("Beans",2000,"Mexico"))

import spark.sqlContext.implicits._
val df = data.toDF("Product","Amount","Country")
df.show()

df 包含三列 产品 、数量和国家

+-------+------+-------+
|Product|Amount|Country|
+-------+------+-------+
| Banana|  1000|    USA|
|Carrots|  1500|    USA|
|  Beans|  1600|    USA|
| Orange|  2000|    USA|
| Orange|  2000|    USA|
| Banana|   400|  China|
|Carrots|  1200|  China|
|  Beans|  1500|  China|
| Orange|  4000|  China|
| Banana|  2000| Canada|
|Carrots|  2000| Canada|
|  Beans|  2000| Mexico|
+-------+-----+-------+

Pivot

pivot 函数旋转将行变成列，执行的是一个聚合的操作。比如上面的例子中，你现在想要统计每个产品在各个国家的数量，实际上是要将国家这个信息由行拓展到列上，这样每行就是统计 每一个产品在各个国家的存储量。

执行的操作如下：

1. 分组：按照产品分组处理
2. 选择需要行转列的对应的 列
3. 执行的聚合操作：这里是sum

val pivotDF = df.groupBy("Product").pivot("Country").sum("Amount")
pivotDF.show()

在行转列的过程中，如果出现没有的数据，那么对应的会被置为 null ，可以使用 null 进行处理填充 0 操作

+-------+------+-----+------+----+
|Product|Canada|China|Mexico| USA|
+-------+------+-----+------+----+
| Orange|  null| 4000|  null|4000|
|  Beans|  null| 1500|  2000|1600|
| Banana|  2000|  400|  null|1000|
|Carrots|  2000| 1200|  null|1500|

Spark 2.0 中函数性能提升

在 1.x 版本执行这个操作算是一个消耗巨大的操作，在 2.0 做出了性能的优化，具体执行的方法可以参考下面的代码实现

val countries = Seq("USA","China","Canada","Mexico")
val pivotDF = df.groupBy("Product").pivot("Country", countries).sum("Amount")
pivotDF.show()

依据 Spark-13749 执行两步聚合达到同样的效果，有进一步的性能优化

val pivotDF = df.groupBy("Product","Country")
      .sum("Amount")
      .groupBy("Product")
      .pivot("Country")
      .sum("sum(Amount)")
pivotDF.show()

Unpivot

​ Unpivot 是一个逆操作，原来是将行转列，现在就是要实现列转行，类似一个stack操作

//unpivot
val unPivotDF = pivotDF.select($"Product",
expr("stack(3, 'Canada', Canada, 'China', China, 'Mexico', Mexico) as (Country,Total)"))
.where("Total is not null")
unPivotDF.show()

+-------+-------+-----+
|Product|Country|Total|
+-------+-------+-----+
| Orange|  China| 4000|
|  Beans|  China| 1500|
|  Beans| Mexico| 2000|
| Banana| Canada| 2000|
| Banana|  China|  400|
|Carrots| Canada| 2000|
|Carrots|  China| 1200|
+-------+-------+-----+