在Spark 2.0中进行列级操作的有效方法

Question

我有两个数据集

数据集1：

id  a    b      c     d
1  0.3  0.1   0.2   0.2
2  0.2  0.3   0.3   0.4
3  0.2  0.4   0.7   0.7
....

dataset2

id  x    
1   8  
2   4 
3   10  
....

我想执行一个操作，在dataSet2中使用dataset1中的每个列中的" x"列，每个ID中的敌人，以便所需的输出为：

id   a    b    c    d
1   2.4   0.8  1.6  1.6
2   0.8   1.2  1.2  1.6
3    2     4    7    7

我所做的是通过加入DataSet2

来映射DataSet1中的每一行

val result = dataset1.join(dataset2, Seq("id")
                     .map(row=> row.getAs[String]("id"),
                          row=> row.getAs[Double]("a") * row.getAs[Int]("x"),
                          row=> row.getAs[Double]("b") * row.getAs[Int]("x"),
                          row=> row.getAs[Double]("c") * row.getAs[Int]("x"),
                          row=> row.getAs[Double]("d") * row.getAs[Int]("x"))

我觉得这样的写作有点多余。有什么方法可以更清楚吗？

Answer 1

您需要的只是select：

dataset1.join(dataset2, Seq("id")).select(
  $"id", $"a" * $"x", $"b" * $"x", $"c" * $"x", $"d" * "x"
).toDF("id", "a", "b", "c", "d")

可以推广

val exprs = $"id" +: dataset1.columns.tail.map(c => (col(c) * $"x").alias(c))
dataset1.join(dataset2, Seq("id")).select(exprs: _*)