델타 데이터의 Spark Dataframe 창 함수 row_number 열 업데이트

Question

델타 데이터에 대한 데이터 프레임의 row_number 열을 업데이트해야합니다.

나는 다음과 같이 기저의 ROW_NUMBER를 구현 한

:

입력 데이터 : 이제

val baseDF = DS1.select(col("KEY1"), col("KEY2"), col("VAL") ,row_number().over(Window.partitionBy(col("KEY1"), col("KEY2")).orderBy(col("KEY1"), col("KEY2").asc)).alias("Row_Num")) 

baseDF.show() 

|KEY1|KEY2|VAL|Row_Num| 
+----+----+---+-------+ 
|001 |a |abc|1  | 
|001 |a |123|2  | 
|002 |b |dfr|1  | 
|003 |c |456|1  | 
|003 |c |ytr|2  | 
+----+----+---+-------+ 

:

val base = List(List("001", "a", "abc"), List("001", "a", "123"),List("003", "c", "456") ,List("002", "b", "dfr"), List("003", "c", "ytr")) 
    .map(row => (row(0), row(1), row(2))) 

    val DS1 = base.toDF("KEY1", "KEY2" ,"VAL") 

    DS1.show() 
+----+----+---+ 
|KEY1|KEY2|VAL| 
+----+----+---+ 
| 001| a|abc| 
| 001| a|123| 
| 003| c|456| 
| 002| b|dfr| 
| 003| c|ytr| 
+----+----+---+ 

는 지금은 아래와 같이 윈도우 함수를 사용하여 ROW_NUMBER를 추가 한 델타로드는 다음과 같습니다.

val delta = List(List("001", "a", "y45") ,List("002", "b", "444")) 
    .map(row => (row(0), row(1), row(2))) 

val DS2 = delta.toDF("KEY1", "KEY2" ,"VAL") 

DS2.show() 

+----+----+---+ 
|KEY1|KEY2|VAL| 
+----+----+---+ 
| 001| a|y45| 
| 002| b|444| 
+----+----+---+ 

그래서 예상되는 업데이트 결과는 dataframes/데이터 세트를 사용하여이 솔루션을 구현할

baseDF.show() 

|KEY1|KEY2|VAL|Row_Num| 
+----+----+---+-------+ 
|001 |a |abc|1  | 
|001 |a |123|2  | 
| 001| a|y45|3  | -----> Delta record 
|002 |b |dfr|1  | 
| 002| b|444|2  | -----> Delta record 
|003 |c |456|1  | 
|003 |c |ytr|2  | 
+----+----+---+-------+ 

어떤 제안해야 하는가? 스파크 rdd의 zipWithIndex로 위의 해결책을 얻을 수 있습니까? 갱신 된 행 번호와 델타 추가

Answer 1

한가지 방법이다 : 다음과 같이, 1)와, DS2으로 다수 2) 조합 baseDF을 열 Row_Num을 추가하고, 3) 새 행 번호를 계산한다

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

val combinedDF = baseDF.union(
    DS2.withColumn("Row_Num", lit(Long.MaxValue)) 
) 

val resultDF = combinedDF.select(
    col("KEY1"), col("KEY2"), col("VAL"), row_number().over(
    Window.partitionBy(col("KEY1"), col("KEY2")).orderBy(col("Row_Num")) 
).alias("New_Row_Num") 
) 

resultDF.show 
+----+----+---+-----------+ 
|KEY1|KEY2|VAL|New_Row_Num| 
+----+----+---+-----------+ 
| 003| c|456|   1| 
| 003| c|ytr|   2| 
| 002| b|dfr|   1| 
| 002| b|444|   2| 
| 001| a|abc|   1| 
| 001| a|123|   2| 
| 001| a|y45|   3| 
+----+----+---+-----------+