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

2023年3月1日 261次阅读

我正在使用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函数时它不起作用的原因.请让我知道如何解决这个问题.

最佳答案 如果要将聚合函数与“普通”列一起使用,则函数应位于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)
点赞