Spark Scala中的列操作

Question

我正在学习使用Apache Spark (Scala)，仍然在弄清楚事情是如何在这里工作的

我正在尝试完成一个简单的任务

查找列的最大值
1. 从这个最大值中减去每个列的值并创建一个新列

我使用的代码是

import org.apache.spark.sql.functions._
val training = sqlContext.createDataFrame(Seq(
  (10),
  (13),
  (14),
  (21)
)).toDF("Values")
val training_max = training.withColumn("Val_Max",training.groupBy().agg(max("Values"))
val training_max_sub = training_max.withColumn("Subs",training_max.groupBy().agg(col("Val_Max")-col("Values) ))

然而，我得到了很多错误。我或多或少精通R，如果我做同样的任务，我的代码会是:

library(dplyr)
new_data <- training %>%
    mutate(Subs= max(Values) - Values)

Answer 1

这是一个使用窗口函数的解决方案。你需要一个HiveContext来使用它们

import org.apache.spark.sql.hive.HiveContext
import org.apache.spark.sql.functions._
import org.apache.spark.sql.expressions.Window
val sqlContext = new HiveContext(sc)
import sqlContext.implicits._
val training = sc.parallelize(Seq(10,13,14,21)).toDF("values")
training.withColumn("subs", 
   max($"values").over(Window.partitionBy()) - $"values").show

产生预期的输出:

+------+----+
|values|subs|
+------+----+
|    10|  11|
|    13|   8|
|    14|   7|
|    21|   0|
+------+----+