How to replace a spark dataframe row with another spark dataframe's row using java

I have 2 data frames

df1

+--------------------+---+--------------------+--------------------+
|                 ID |B  |C                   |            D       |
+--------------------+---+--------------------+--------------------+
|                   1|1.0|                 1.0|                 1.0|
|                   2|2.0|                 2.0|                 2.0|
|                   3|3.0|                 3.0|                 3.0|
|                   4|4.0|                 4.0|                 4.0|
+--------------------+---+--------------------+--------------------+

df2

+--------------------+---+--------------------+--------------------+
|                 ID |B  |C                   |            D       |
+--------------------+---+--------------------+--------------------+
|                   1|100|                 1.0|                 100|
+--------------------+---+--------------------+--------------------+

If ID in df2 matches an ID in df1, I want to replace the row in df1 with the updated values in df2. So the new df1 looks like:

df1

+--------------------+---+--------------------+--------------------+
|                 ID |B  |C                   |            D       |
+--------------------+---+--------------------+--------------------+
|                   1|100|                 1.0|                 100|
|                   2|2.0|                 2.0|                 2.0|
|                   3|3.0|                 3.0|                 3.0|
|                   4|4.0|                 4.0|                 4.0|
+--------------------+---+--------------------+--------------------+

I've been trying to figure this out with union and join and just not having any luck yet. I first created a new dataframe based on filtering for the ID of df1 and that works and I called that dataframe matchedDF that looks like:

matchedDF (dataframe based on finding a match of ID 1 in df1)

+--------------------+---+--------------------+--------------------+
|                 ID |B  |C                   |            D       |
+--------------------+---+--------------------+--------------------+
|                   1|1.0|                 1.0|                 1.0|
+--------------------+---+--------------------+--------------------+

But I don't know if I just want to delete the original ID 1 in df1 and add the new matchedDF or do I somehow want to update the original ID 1 with the matchedDf? Or am I approaching this all wrong?

Thanks

To stay computationally efficient, it's always a good idea to avoid joins/shuffles where possible.

This looks like a case where it is possible to avoid joining, have a look at the following code (it is in Scala, but the principles remain the same):

// Constructing the 2 dfs
val df = Seq(
  (1, 1.0, 1.0, 1.0),
  (2, 2.0, 2.0, 2.0),
  (3, 3.0, 3.0, 3.0),
  (4, 4.0, 4.0, 4.0)
).toDF("ID", "B", "C", "D")

val df2 = Seq(
  (1, 100, 1.0, 100),
  (2, 100, 1.0, 100)
).toDF("ID", "B", "C", "D")

// Collecting the IDs to be updated into a single Array
// IMPORTANT: we make the assumption that this array is not large (in your
// example there is only 1 row here, so the array only has 1 element which is
// totally fine)
val newIds = df2.select("ID").collect.map(_.getInt(0))

// Removing the original rows with the unwanted IDs and unioning the result with
// the new rows
val output = df
  .filter(not(col("ID").isin(newIds: _*)))
  .union(df2)

scala> output.show
+---+-----+---+-----+                                                           
| ID|    B|  C|    D|
+---+-----+---+-----+
|  3|  3.0|3.0|  3.0|
|  4|  4.0|4.0|  4.0|
|  1|100.0|1.0|100.0|
|  2|100.0|1.0|100.0|
+---+-----+---+-----+

So basically, if we can make the assumption that df2 (with the new values) is small like in your example, you can do something like the following:

• collect the ID values into a single (undistributed) Array. From your example it seems like this is OK. If the amount of new rows is really large this might not be the best approach
• filter the original df using the isin method of a column and negating using not (basically removing the rows with the new IDs)
• union the filtered df and df2, resulting in the rows being updated WITHOUT any expensive operation like a shuffle