[Python] How to re-order and upsert records after combining two dataframes

I have below similar scenarios.

I have a source table which doesn't contain an id and I need to merge the records from the source into the target table which contains id which is consistent in the order.

SourceDF

+-------+---------+-------+----------------+-----------+-------+----------------+
|s_gp | s_sys | s_d |s_fptr |s_rgp | s_flag| s_line |
+-------+---------+-------+----------------+-----------+-------+----------------+
| abc|s_system1|s_feed1| s_para_1| s_run_gp1|sflag_1| s_pipe1|
| def|s_system3|s_feed3| s_para_3| s_run_gp3|sflag_3| s_pipe3|
| bcz|s_system2|s_feed2| s_para_2| s_run_gp2|sflag_2|s_pipe2_upadated|
| fjk|s_system4|s_feed4| s_para_4| s_run_gp4|sflag_4| s_pipe4|
| dfg|s_system5|s_feed5| s_para_5| s_run_gp5|sflag_5| s_pipe5|
+-------+---------+-------+----------------+-----------+-------+----------------+


TargetDF

+---+-------+---------+-------+----------------+-----------+-------+----------+
| id|s_gp | s_sys | s_d | s_fptr | s_rgp | s_flag| s_line |
+---+-------+---------+-------+----------------+-----------+-------+----------+
|101| abc|s_system1|s_feed1| s_para_1| s_run_gp1|sflag_1| s_pipe1|
|205| bcz|s_system2|s_feed2| s_para_2| s_run_gp2|sflag_2| s_pipe2|
|103| def|s_system3|s_feed3| s_para_3| s_run_gp3|sflag_3| s_pipe3|
+---+-------+---------+-------+----------------+-----------+-------+----------+


Here I want to insert the new record from the SourceDF with the incremental id. If the records are updated, then I need to update in the target table. The id in the TargetDF is consistent. So if records in the SourceDF are wrongly placed, then it should re-order and update in the target table.

The case is:

1. Record with s_gp - "bcz" which stands with id - 205 as per TargetDF has wrongly placed in SourceDF. But with updated value "s_pipe2_upadated", so when updating into the target table, the order should be corrected - i.e. it should update against 205 only.
2. If it is a completely new record in SourceDF as per TargetDF, then it should insert as new record.

join_cols = [col for col in source_df.columns if col in target_df.columns and col != "id"]
print("Join columns", join_cols)
matched_df = source_df.join(
target_df.select("id", *join_cols),
on=join_cols,
how="left"
)
existing_records_df = matched_df.filter(F.col("id").isNotNull())
print("existing_records_df")
display(existing_records_df)

new_records_df = matched_df.filter(F.col("id").isNull() | (F.col("id").isNotNull() & (F.concat_ws(",", *[F.col(col) for col in join_cols]) != F.concat_ws(",", *[F.col(f"target_df.{col}") for col in join_cols]))))
print("new_records_df")
display(new_records_df)

print("new_records_df")
display(new_records_df)
max_id = target_df.select(F.max("id")).first()[0] or 0

new_records_with_ids = new_records_df.withColumn(
"id", F.row_number().over(Window.orderBy(F.lit(1))) + max_id
)
print("new_records_with_ids")
display(new_records_with_ids)
combined_df = existing_records_df.unionByName(new_records_with_ids)
print("combined")
display(combined_df)


The above code/approach is not the right solution as I tested. I am expecting below expected TargetDF.

Final TargetDF after upsert

+---+-------+---------+-------+----------------+-----------+-------+-----------------+
| id|s_gp | s_sys | s_d | s_fptr | s_rgp | s_flag| s_line |
+---+-------+---------+-------+----------------+-----------+-------+-----------------+
|101| abc|s_system1|s_feed1| s_para_1| s_run_gp1|sflag_1| s_pipe1 |
|205| bcz|s_system2|s_feed2| s_para_2| s_run_gp2|sflag_2| s_pipe2_upadated|
|103| def|s_system3|s_feed3| s_para_3| s_run_gp3|sflag_3| s_pipe3 |
|104| fjk|s_system4|s_feed4| s_para_4| s_run_gp4|sflag_4| s_pipe4 |
|105| dfg|s_system5|s_feed5| s_para_5| s_run_gp5|sflag_5| s_pipe5 |
+---+-------+---------+-------+----------------+-----------+-------+----------+------+

Continue reading...