嗨,我有两张这样的桌子.
源表
orig1 orig2 orig3 xref1 xref2 xref3
1 1 1 2 2 2
1 1 1 3 3 3
23 23 23 12 12 12
目标表：
orig1 orig2 orig3 xref1 xref2 xref3 version
1 1 1 1 1 1 0
我需要输出如下
1)我需要匹配(源(orig1 orig2 orig3)== target(orig1 orig2 orig3)),
如果它的macthing我们需要通过将版本增加1来从源附加到目标表
如果它不匹配,只需将版本附加为’0′
预期产量是：
orig1 orig2 orig3 xref1 xref2 xref3 version
1 1 1 1 1 1 0
1 1 1 2 2 2 1
1 1 1 3 3 3 2
23 23 23 12 12 12 0
我试过数据帧级别.但它没有按预期工作.任何帮助,将不胜感激.
我尝试了以下方式.
val source = spark.sql("select xref1,xref2,xref3,orig1,orig2,orig3 from default.source")
val target = spark.sql("select xref1,orig3 from default.target")
val target10 = spark.sql("select xref1,orig3,version from default.target")
val diff=( source.select("xref1","xref2","xref3","orig1","orig2","orig3") == target.select("xref1","orig3"))
if ( diff == false ){
val diff1 = source.select("orig1","orig3").except(target.select("orig1","orig3"))
if ( diff1.count > 0 ) {
val ver = target10.groupBy("orig1","orig3").max("version")
val common = source.select("orig1","orig3").intersect(target.select("orig1","orig3"))
val result = common.join(ver,common("orig1") === ver("orig1") && common("orig2") === ver("orig2") && common("orig3") === ver("orig3"),"inner").select(ver("orig1"),ver("orig2"),ver("orig3"),(ver("max(version)") + 1
) as "version")
val result1 = result.join(source,result("orig1") === source("orig1") && result("orig2") === source("orig2") && result("orig3") === source("orig3"),"inner").select(source("orig1"),source("orig2"),source("orig3"),result("version"),source("xref1"),source("xref2"),source("xref3"))
val result2=source.select("orig1","orig3")).withColumn("version",lit(0))
val execpettarget=result2.select($"orig1".alias("DIV"),$"orig2".alias("SEC"),$"orig3".alias("UN"),$"version".alias("VER"))
val result23 = execpettarget.join(source,execpettarget("DIV") === source("orig1") && execpettarget("SEC") === source("orig2") && execpettarget("UN") === source("orig3"),execpettarget("VER"),source("orig1"),source("orig3"))
val final_result = result1.union(result23)
final_result.show()
}else{
println("else")
val ver1 = target10.groupBy("orig1","orig3").max("version")
val common1 = source.select("orig1","orig3"))
val result11 = common1.join(ver1,common1("orig1") === ver1("orig1") && common1("orig2") === ver1("orig2") && common1("orig3") === ver1("orig3"),"inner").select(ver1("orig1"),ver1("orig2"),ver1("orig3"),(ver1("max(version)") + 1) as "version")
val result3 = result11.join(source,result11("orig1") === source("orig1") && result11("orig2") === source("orig2") && result11("orig3") === source("orig3"),result11("version"),source("xref3"))
result3.show()
}}
但是在最后的连接中,源有2个重复的行.因此,当使用目标加入源代码时,我会获得多行.
最佳答案
我没有尝试破译你的代码,但是基于源代码,目标和代码.预期的结果我认为这可能是一个解决方案：
val w = Window.partitionBy('orig1,'orig2,'orig3).orderBy('version.desc)
val joined = source
.withColumn("version",lit(null).cast(IntegerType))
.union(target)
.withColumn("version",row_number().over(w) + coalesce(max('version).over(w),lit(0)) - lit(1))
joined.show()
我的想法是,连接没有意义,因为您希望最终得到来自目标的来源#记录的#记录：=>联盟
在此联合之后,您希望处理每组相似的键(orig1,orig3)=>窗口
您关心组中的最高版本号,否则选择0：=>最大和最大合并
您希望将此max作为offet用于窗口其余部分的排名：=> ROW_NUMBER
+-----+-----+-----+-----+-----+-----+-------+
|orig1|orig2|orig3|xref1|xref2|xref3|version|
+-----+-----+-----+-----+-----+-----+-------+
| 23| 23| 23| 12| 12| 12| 0|
| 1| 1| 1| 1| 1| 1| 0|
| 1| 1| 1| 2| 2| 2| 1|
| 1| 1| 1| 3| 3| 3| 2|
+-----+-----+-----+-----+-----+-----+-------+