加入后如何更新 Spark Scala 中的数据框列？答案

【问题标题】：How to update dataframe column in Spark Scala after join?加入后如何更新 Spark Scala 中的数据框列？
【发布时间】：2018-09-06 03:22:03
【问题描述】：

将两个数据框合并成近 60 列。他们中的大多数人假设保持原样，但有些人需要根据其他列中的值进行更新。有没有办法在不计算新列的情况下更新这些列、删除原始列并重命名计算后的列？

简化示例：左侧数据框的$"Sales 列中的收入应该由连接结果中的$"Weight 加权。有没有一种有效的方法可以在不生成$"SalesWeighted 作为新列的情况下进行计算，将原始$Sales 删除并将$SalesWeighted 重新命名为$Sales？

val l = Seq((1, 50), (2, 35), (3, 66))
            .toDF("Id", "Sales")

val r = Seq((1, "Premium", 0.2), (1, "Standard", 0.8), 
            (2, "Premium", 0.4), (2, "Standard", 0.6), 
            (3, "Premium", 0.333), (3, "Standard", 0.333), (3, "Garbage", 0.334))
            .toDF("Id", "Grade", "Weight")

display(l.join(r, Seq("Id")).withColumn("SalesWeighted", $"Sales"*$"Weight")
            .orderBy($"Id", $"Grade"))

【问题讨论】：

标签： scala apache-spark apache-spark-sql

【解决方案1】：

您可以简单地将新列命名为与要替换的列相同的名称：

l.join(r, Seq("Id")).withColumn("Sales", $"Sales" * $"Weight").
  orderBy($"Id", $"Grade")

或者，只需使用select：

l.join(r, Seq("Id")).
  select($"Id", $"Grade", $"Weight", ($"Sales" * $"Weight").as("Sales")).
  orderBy($"Id", $"Grade")

【讨论】：

我没有意识到withColumn 替换原来的专栏！这正是我正在寻找的解决方案。 select 也会生成答案，但有两个缺点：现实生活场景需要手动列出许多列，并且任何上游更改也需要编辑这部分代码。

【解决方案2】：

使用Drop删除不必要的列

val l = Seq((1, 50), (2, 35), (3, 66))
                .toDF("Id", "Sales")

    val r = Seq((1, "Premium", 0.2), (1, "Standard", 0.8), 
                (2, "Premium", 0.4), (2, "Standard", 0.6), 
                (3, "Premium", 0.333), (3, "Standard", 0.333), (3, "Garbage", 0.334))
                .toDF("Id", "Grade", "Weight")

    display(l.join(r, Seq("Id")).withColumn("SalesWeighted", $"Sales"*$"Weight").drop($"Sales")
                .orderBy($"Id", $"Grade"))

【讨论】：

是的，结合drop 和withColumnRenamed 效果很好：display(l.join(r, Seq("Id")).withColumn("SalesWeighted", $"Sales"*$"Weight").drop($"Sales").withColumnRenamed("SalesWeighted", "Sales").orderBy($"Id", $"Grade")) 我只是想知道是否有办法一步而不是3（类似于R 数据帧操作）。