Pyspark - отдельные записи на основе 2 столбцов в датафрейме

У меня есть 2 кадра данных, например df1 и df2.

Данные df1 поступают из базы данных, а df2 - это новые данные, которые я получаю от своего клиента. Мне нужно обработать новые данные и выполнить UPSERTs зависимости от того, UPSERTs ли новая запись или существующая запись.

Пример вывода данных:

df1= sqlContext.createDataFrame([("xxx1","81A01","TERR NAME 01","NJ"),("xxx2","81A01","TERR NAME 01","NJ"),("xxx3","81A01","TERR NAME 01","NJ"),("xxx4","81A01","TERR NAME 01","CA"),("xx5","81A01","TERR NAME 01","ME")], ["zip_code","territory_code","territory_name","state"])
df2= sqlContext.createDataFrame([("xxx1","81A01","TERR NAME 55","NY"),("xxx2","81A01","TERR NAME 55","NY"),("x103","81A01","TERR NAME 01","NJ")], ["zip_code","territory_code","territory_name","state"])

df1.show()
+--------+--------------+--------------+-----+
|zip_code|territory_code|territory_name|state|
+--------+--------------+--------------+-----+
|    xxx1|         81A01|  TERR NAME 01|   NJ|
|    xxx2|         81A01|  TERR NAME 01|   NJ|
|    xxx3|         81A01|  TERR NAME 01|   NJ|
|    xxx4|         81A01|  TERR NAME 01|   CA|
|    xxx5|         81A01|  TERR NAME 01|   ME|
+---------------------------------------------

# Print out information about this data
df2.show()
+--------+--------------+--------------+-----+
|zip_code|territory_code|territory_name|state|
+--------+--------------+--------------+-----+
|    xxx1|         81A01|  TERR NAME 55|   NY|
|    xxx2|         81A01|  TERR NAME 55|   NY|
|    x103|         81A01|  TERR NAME 01|   NJ|
+---------------------------------------------

Ожидаемые результаты: Мне нужно сравнить df2 с фреймом df1. Создайте 2 новых набора данных, основанных на приведенном выше сравнении, то есть обновляемые записи и записи, которые необходимо добавить/добавить в базу данных.

если zip_code & territory_code одинаковы, то это UPDATE, иначе это ВСТАВИТЬ в базу данных.

Например: Новый вывод dataframe для INSERT:

 +--------+--------------+--------------+-----+
 |zip_code|territory_code|territory_name|state|
 +--------+--------------+--------------+-----+
 |    x103|         81A01|  TERR NAME 01|   NJ|
 +---------------------------------------------

Новый фрейм данных для UPDATE:

+--------+--------------+--------------+-----+
|zip_code|territory_code|territory_name|state|
+--------+--------------+--------------+-----+
|    xxx1|         81A01|  TERR NAME 55|   NY|
|    xxx2|         81A01|  TERR NAME 55|   NY|
+---------------------------------------------

Может кто-нибудь, пожалуйста, помогите мне? И я использую AWS Glue.

ОБНОВЛЕНИЕ: РЕШЕНИЕ (Использование объединения и вычитания)

df3 = df1.join(df2, (df1.zip_code == df2.zip_code_new) & (df1.territory_code == df2.territory_code_new))
df5=df3.drop("zip_code", "territory_code", "territory_name", "state")
df5.show()

+------------+------------------+------------------+---------+
|zip_code_new|territory_code_new|territory_name_new|state_new|
+------------+------------------+------------------+---------+
|        x103|             81A01|      TERR NAME 01|       NJ|
+------------+------------------+------------------+---------+

df4=df2.subtract(df5)
df4.show()

+------------+------------------+------------------+---------+
|zip_code_new|territory_code_new|territory_name_new|state_new|
+------------+------------------+------------------+---------+
|    xxx1    |         81A01    |  TERR NAME 55    |   NY    |
|    xxx2    |         81A01    |  TERR NAME 55    |   NY    |
+------------------------------------------------------------+

Для обновления базы данных RDS я использую pymysql/Mysqldb:

db = MySQLdb.connect("xxxx.rds.amazonaws.com", "username", "password", "databasename")
cursor = db.cursor()

#cursor.execute("REPLACE INTO table SELECT * FROM table_stg")
insertQry = "INSERT INTO table VALUES('xxx1','81A01','TERR NAME 55','NY') ON DUPLICATE KEY UPDATE territory_name='TERR NAME 55', state='NY'"
n=cursor.execute(insertQry)
db.commit()
cursor.fetchall()
db.close()

Спасибо