PySpark: Insert or update dataframe with another dataframe

Question

I have two dataframes, DF1 and DF2. DF1 is the master and DF2 is the delta. The data from DF2 should be inserted into DF1 or used to update the DF1 data.

Lets say the DF1 is of the following format:

id_no	start_date	amount	days
1	2016-01-01	4650	22
2	2016-01-02	3130	45
1	2016-01-03	4456	22
2	2016-01-15	1234	45

DF2 contains the following:

id_no	start_date	amount	days
1	2016-01-01	8650	52
2	2016-01-02	7130	65
1	2016-01-06	3456	20
2	2016-01-20	2345	19
3	2016-02-02	1345	19

I need to combine the two dataframes such that if the "id_no" and "start date" of DF2 matches DF1, it should be replaced in DF1 and if does not match, it should be inserted into DF1. The "id_no" is not unique.

The expected result:

id_no	start_date	amount	days
1	2016-01-01	8650	52
2	2016-01-02	7130	65
1	2016-01-03	4456	22
2	2016-01-15	1234	45
1	2016-01-06	3456	20
2	2016-01-20	2345	19
3	2016-02-02	1345	19

Answer 1

You can join the two data frames on id_no and start_date, and then coalesce the amount and days column with columns from df2 coming first:

import pyspark.sql.functions as f

df1.alias('a').join(
    df2.alias('b'), ['id_no', 'start_date'], how='outer'
).select('id_no', 'start_date', 
    f.coalesce('b.amount', 'a.amount').alias('amount'), 
    f.coalesce('b.days', 'a.days').alias('days')
).show()

+-----+----------+------+----+
|id_no|start_date|amount|days|
+-----+----------+------+----+
|    1|2016-01-06|  3456|  20|
|    2|2016-01-20|  2345|  19|
|    1|2016-01-03|  4456|  22|
|    3|2016-02-02|  1345|  19|
|    2|2016-01-15|  1234|  45|
|    1|2016-01-01|  8650|  52|
|    2|2016-01-02|  7130|  65|
+-----+----------+------+----+

If you have many more columns:

cols = ['amount', 'days']

df1.alias('a').join(
    df2.alias('b'), ['id_no', 'start_date'], how='outer'
).select('id_no', 'start_date', 
    *(f.coalesce('b.' + col, 'a.' + col).alias(col) for col in cols)
).show()
+-----+----------+------+----+
|id_no|start_date|amount|days|
+-----+----------+------+----+
|    1|2016-01-06|  3456|  20|
|    2|2016-01-20|  2345|  19|
|    1|2016-01-03|  4456|  22|
|    3|2016-02-02|  1345|  19|
|    2|2016-01-15|  1234|  45|
|    1|2016-01-01|  8650|  52|
|    2|2016-01-02|  7130|  65|
+-----+----------+------+----+

Answer 2

union and subsequent agg would work.

from pyspark.sql import functions as F

grp_by = {'id_no', 'start_date'}
df = df1.union(df2)
df = df.groupby(*grp_by).agg(*[F.first(c).alias(c) for c in set(df.columns)-grp_by])

df.show()
# +-----+----------+------+----+
# |id_no|start_date|amount|days|
# +-----+----------+------+----+
# |    2|2016-01-02|  3130|  45|
# |    1|2016-01-01|  4650|  22|
# |    1|2016-01-03|  4456|  22|
# |    2|2016-01-15|  1234|  45|
# |    3|2016-02-02|  1345|  19|
# |    1|2016-01-06|  3456|  20|
# |    2|2016-01-20|  2345|  19|
# +-----+----------+------+----+