如何解决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|
+----+----+----+----+----+
版权声明:本文内容由互联网用户自发贡献,该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 dio@foxmail.com 举报,一经查实,本站将立刻删除。