Сравнение двух строк на уровне строк

Question

Привет, у меня есть две такие таблицы.

исходная таблица

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) Мне нужно соответствовать (source(orig1 orig2 orig3) == target(orig1 orig2 orig3)), если это происходит, нам нужно добавить исходную таблицу к целевой, увеличив версию на 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,xref2,xref3,orig1,orig2,orig3 from default.target")
val target10 = spark.sql("select xref1,xref2,xref3,orig1,orig2,orig3,version from default.target")
val diff=( source.select("xref1","xref2","xref3","orig1","orig2","orig3") == target.select("xref1","xref2","xref3","orig1","orig2","orig3"))
if  ( diff == false ){
  val diff1 = source.select("orig1","orig2","orig3").except(target.select("orig1","orig2","orig3"))
  if (  diff1.count > 0 ) {
   val ver = target10.groupBy("orig1","orig2","orig3").max("version")
   val common = source.select("orig1","orig2","orig3").intersect(target.select("orig1","orig2","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","orig2","orig3").except(target.select("orig1","orig2","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"), "inner").select(source("orig1"),source("orig2"),source("orig3"),execpettarget("VER"),source("orig1"),source("orig2"),   source("orig3"))
   val final_result = result1.union(result23)
   final_result.show()
}else{
println("else")
val ver1 = target10.groupBy("orig1","orig2","orig3").max("version")
val common1 = source.select("orig1","orig2","orig3").intersect(target.select("orig1","orig2","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"), "inner").select(source("orig1"),source("orig2"),source("orig3"),result11("version"),source("xref1"),source("xref2"),source("xref3"))
result3.show()
}}

Но в финальном соединении у источника есть 2 повторяющихся строки. Поэтому при соединении источника с целью я получаю несколько строк.

Answer 1

Я не пытался расшифровать ваш код, но исходя из исходного, целевого и ожидаемого результата, я бы сказал, что это может быть решением:

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()

Моя идея заключалась в том, что объединение не имеет смысла, потому что вы хотите получить # записей из источника + # записей из цели: => union

После этого объединения вы хотите обработать каждую группу похожих ключей (orig1, orig2, orig3) => Window

Вас интересует самый высокий номер версии в группе, в противном случае выберите 0: => max & coalesce

Вы хотите использовать этот максимум в качестве оффета для ранжирования остальной части окна: => 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|
+-----+-----+-----+-----+-----+-----+-------+