Spark SQL：从一个数据帧的相应行中减去另一行

问题描述

我有2个具有相同架构的数据框

df1
    col1   col2
    23     89
df2
    col1   col2
    57     72

我想逐行从df2中减去df1。所以我想看

result
col1   col2
34     -3

如何在Spark sql中实现此目标？

解决方法

以下代码可能会有所帮助，

import org.apache.spark.sql.expressions.Window

val df1 = Seq((23,89)).toDF("col1","col2")

val df2 = Seq((57,72)).toDF("col1","col2")

val windowSpec  = Window.partitionBy(lit("A")).orderBy(lit("A"))

val df3=df1.withColumn("id",row_number.over(windowSpec))
val df4=df2.withColumn("id",row_number.over(windowSpec))

df3.createOrReplaceTempView("df3")
df4.createOrReplaceTempView("df4")

spark.sql("SELECT a.col1-b.col1 as col1,a.col2-b.col2 as col2 FROM df4 a INNER JOIN df3 b ON a.id=b.id").show()

/*
+----+----+
|col1|col2|
+----+----+
|  34| -17|
+----+----+
*/

另一种更困难的方法，但可能是明智的性能，严格来说是正确的。显示了DF API常用名称的一些问题。

zipWithIndex方法可保留顺序。

要点仍然是，如果要几百个cols呢？

SELECT 
   Date_FY
FROM 
   Date_Lookup_Tbl_FY
WHERE
   Date_FY IN (SELECT Date_FY
               FROM Date_Lookup_Tbl_FY
               WHERE Date_FY >= Max(Date_FY) - 1
              )

附录

对于许多列来说，这只是一个开始，而不是答案的一部分：

import org.apache.spark.sql.DataFrame
import org.apache.spark.sql.Row
import org.apache.spark.sql.types.{StructField,StructType,IntegerType,ArrayType,LongType}

def renameCols(t: DataFrame,suffix: String = ""): DataFrame = {
  t.select( t.columns.map { c => t.col(c).as( c + suffix) } : _* )
}

// Initial data
val df1 = sc.parallelize(Seq( (1.0,2.0),(4.0,2.0))).toDF("c1","c2")
val df2 = sc.parallelize(Seq( (1.0,3.0),(1.0,"c2")

val newSchema = StructType(df1.schema.fields ++ Array(StructField("rowId",LongType,false)))

val rddWithId1 = df1.rdd.zipWithIndex
val rddWithId2 = df2.rdd.zipWithIndex
val X = spark.createDataFrame(rddWithId1.map{ case (row,index) => Row.fromSeq(row.toSeq ++ Array(index))},newSchema) 
val Y = spark.createDataFrame(rddWithId2.map{ case (row,newSchema) 
val dfA = renameCols(X,"_1")
val dfB = renameCols(Y,"_2")

val df = dfA.join(dfB,dfA("rowId_1") === dfB("rowId_2")) 
df.show(false)
df.selectExpr("c1_1 - c1_2 as c1","c2_1 - c2_2 as c2").show(false)

通过点亮和替换，并删除rowId将适当的功能应用于_1和_2列。仔细考虑的好例子。

此版本针对DF中可变个列进行此操作。只需将val n更新为DF中的列数即可。

它回答了发问者随后提出的询问，这不是原始问题的一部分。

import org.apache.spark.sql.DataFrame
import org.apache.spark.sql.Row
import org.apache.spark.sql.types.{StructField,LongType}
import spark.implicits._
import org.apache.spark.sql.functions._
import org.apache.spark.sql.Column

val suffix = "_2"

def renameCols(t: DataFrame,suffix: String = ""): DataFrame = {
  t.select( t.columns.map { c => t.col(c).as( c + suffix) } : _* )
}

// Initial data
val df1 = sc.parallelize(Seq( (11.0,22.0,33.0,44.0,55.0),(22.0,55.0,66.0))).toDF("c1","c2","c3","c4","c5")
val df2 = sc.parallelize(Seq( (1.0,2.0,3.0,4.0,5.0),(2.0,5.0,6.0))).toDF("c1","c5")
val newSchema = StructType(df1.schema.fields ++ Array(StructField("rowId","_1_2")

val df = dfA.join(dfB,dfA("rowId_1") === dfB("rowId_1_2")) 
//df.show(false)

val res = df.select(df.columns.filter(_.endsWith("_1")).map(c => col(c) - col(c+s"${suffix}")): _*).drop("(rowId_1 - rowId_1_2)")

// Number of cols generated,could alo be automated
val n = 5
val newColumns = Seq.range(1,n+1).map(c => ("c_" + c))
res.toDF(newColumns:_*).show(false)

+----+----+----+----+----+
|c_1 |c_2 |c_3 |c_4 |c_5 |
+----+----+----+----+----+
|10.0|20.0|30.0|40.0|50.0|
|20.0|30.0|40.0|50.0|60.0|
+----+----+----+----+----+

apache-spark apache-spark-sql