spark中的Dataframe比较：scala

ikfrs5lh 于 2021-05-27 发布在 Spark

关注(0)|答案(1)|浏览(521)

我在spark/scala中有两个数据框，其中有一些公共列，如工资、奖金、增量等。我需要比较这两个数据框的列，并且任何变化，如第一个数据框的工资是3000，第二个数据框的工资是5000，然后我需要在新的数据框中插入5000-3000=2000作为工资，如果第一个数据框中的salary是5000，第二个数据框中salary是3000，那么我需要在新的数据框中插入5000+3000=8000作为salary，如果两个数据框中的salary相同，那么需要从第二个数据框中插入。

val columns = df1.schema.fields.map(_.salary)
    val salaryDifferences = columns.map(col => df1.select(col).except(df2.select(col)))
    salaryDifferences.map(diff => {if(diff.count > 0) diff.show})

我试过上面的查询，但是它的给定列和值有什么不同，我还需要检查diff是负的还是正的，基于这个我需要执行逻辑。谁能给我一个提示，我如何实现这一点，并在第三个Dataframe中插入记录，

apache-spark apache-spark-sql

来源：https://stackoverflow.com/questions/61792472/dataframe-comparision-in-spark-scala

1条答案

按热度按时间

eblbsuwk1#

连接Dataframe并使用嵌套 when 以及 otherwise 条款。也可以在代码中找到注解

import org.apache.spark.sql.functions._

object SalaryDiff {

  def main(args: Array[String]): Unit = {
    val spark = Constant.getSparkSess
    import spark.implicits._

    val df1 = List(("1", "5000"), ("2", "3000"), ("3", "5000")).toDF("id", "salary") // First dataframe

    val df2 = List(("1", "3000"), ("2", "5000"), ("3", "5000")).toDF("id", "salary") // Second dataframe

    val df3 = df1   // Is your 3rd tables
      .join(
        df2
        , df1("id") === df2("id")    // Join both dataframes on id column
      ).withColumn("finalSalary", when(df1("salary") < df2("salary"), df2("salary") - df1("salary")) // 5000-3000=2000 check
      .otherwise(
      when(df1("salary") > df2("salary"), df1("salary") + df2("salary"))  // 5000+3000=8000  check
        .otherwise(df2("salary"))))    // insert from second dataframe
      .drop(df1("salary"))
      .drop(df2("salary"))
      .withColumnRenamed("finalSalary","salary")
      .show()

  }

}

赞(0）回复(0）举报 2021-05-27

我来回答

spark中的Dataframe比较：scala

1条答案

相关问题

热门标签

最新问答