Spark Scala 中的列操作

Question

我正在学习使用 Apache Spark (Scala)，并且仍在弄清楚这里的工作方式

我正在努力实现一个简单的任务

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

我使用的代码是

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|
+------+----+