scala - 使用 Spark 访问数据框中的列

Question

我正在使用 SCALA 开发 SPARK 1.6.1 版本并面临一个不寻常的问题。使用在同一执行期间创建的现有列创建新列时，会出现“org.apache.spark.sql.AnalysisException”。
工作: .

 val resultDataFrame = dataFrame.withColumn("FirstColumn",lit(2021)).withColumn("SecondColumn",when($"FirstColumn" - 2021 === 0, 1).otherwise(10))
    resultDataFrame.printSchema().

不工作

val resultDataFrame = dataFrame.withColumn("FirstColumn",lit(2021)).withColumn("SecondColumn",when($"FirstColumn" - **max($"FirstColumn")** === 0, 1).otherwise(10))
resultDataFrame.printSchema().

在这里，我使用在同一执行期间创建的 FirstColumn 创建我的 SecondColumn。问题是为什么它在使用 avg/max 函数时不起作用。请让我知道我该如何解决这个问题。

Answer 1

如果您想将聚合函数与“普通”列一起使用，这些函数应该在 groupBy 之后。或使用 Window 定义子句。在这些情况下，它们毫无意义。例子:

val result = df.groupBy($"col1").max("col2").as("max") // This works

在上述情况下，生成的 DataFrame 将同时具有“col1”和“max”作为列。

val max = df.select(min("col2"), max("col2")) 

这是有效的，因为查询中只有聚合函数。但是，以下操作不起作用:

val result = df.filter($"col1" === max($"col2"))

因为我试图将非聚合列与聚合列混合。

如果要将列与聚合值进行比较，可以尝试连接:

val maxDf = df.select(max("col2").as("maxValue"))
val joined = df.join(maxDf)
val result = joined.filter($"col1" === $"maxValue").drop("maxValue")

或者甚至使用简单的值:

val maxValue = df.select(max("col2")).first.get(0)
val result = filter($"col1" === maxValue)