Сравните два Dataframe и запустите «Обновить остальную вставку» в Pyspark

Question

Я новичок в Python и не смог найти точные ответы, которые я ищу в других блогах и, следовательно, опубликовать его как новый вопрос.

У меня есть два кадра данных, созданные как показано ниже

df_hive.show() 
+--------+----------+-------+ 
|BATCH_ID|SRC_SYS_ID|ACT_IND| 
+--------+----------+-------+ 
|     100|      SYS1|      N| 
|     101|      SYS2|      N| 
|     102|      SYS3|      N| 
|     103|      SYS4|      Y| 
+--------+----------+-------+

df_orc.show() 
+--------+----------+-------+ 
|BATCH_ID|SRC_SYS_ID|ACT_IND| 
+--------+----------+-------+ 
|      99|      SYS0|      N|
|     100|      SYS1|      N| 
|     101|      SYS2|      N| 
|     102|      SYS3|      Y| 
+--------+----------+-------+

Ожидаемые результаты для "df_orc"

+--------+----------+-------+ 
|BATCH_ID|SRC_SYS_ID|ACT_IND| 
+--------+----------+-------+ 
|      99|      SYS0|      N|
|     100|      SYS1|      N| 
|     101|      SYS2|      N| 
|     102|      SYS3|      N| 
|     103|      SYS4|      Y| 
+--------+----------+-------+

что я пытаюсь добиться, это соединить два кадра данных на основе BATCH_ID и SRC_SYS_ID и, если они совпадают (и ACT_IND отличается), обновить ACT_IND во втором кадре данных "df_orc" и, если совпадений не найдено, вставить его как новую запись в "df_orc".

Ценю вашу помощь заранее

Answer 1

valuesA = [(100,'SYS1','N'),(101,'SYS2','N'),(102,'SYS3','N'),(103,'SYS4','Y')]
TableA = spark.createDataFrame(valuesA,['BATCH_ID','SRC_SYS_ID','ACT_IND'])

valuesB = [(99,'SYS0','N'),(100,'SYS1','N'),(101,'SYS2','N'),(102,'SYS3','Y')]
TableB = spark.createDataFrame(valuesB,['BATCH_ID','SRC_SYS_ID','ACT_IND'])

ta = TableA.alias('ta')
tb = TableB.alias('tb')

diff = (ta.select('BATCH_ID')
        .subtract(tb.select('BATCH_ID'))
        .rdd.map(lambda x: x[0]).collect())

tb.unionAll(ta[ta.BATCH_ID.isin(diff)]).orderBy('BATCH_ID').show()