Using Update on Delta table is changing the state of an intermediate DataFrame

Question

I am facing a situation here. So below are the steps I am using to transform my DataFrame

val filteringRecordsToExpire = collectAllActiveRecords.join(collectingSrcSysIdsToExpire, Seq("trans_id"), "leftsemi") 

filteringRecordsToExpire contains few of the IDS which I need to make Invalid

val expiredList = filteringRecordsToExpire.select("trans_id").distinct().collect()

expiredList.foreach(v => expireRecords(v(0).toString)) --> Here I am Updating each record

Now I want to use those Same IDs that I expired and further want to re-Enter them in the same Table with some new Values.

But I am getting an Empty DataFrame After I perform the Expire ( Which is basically updating the existing table for those same IDs ) collectingSrcSysIdsToExpire - So this DataFrame holds all those IDs which further I want to modify and INSERT into the Table. But in this process The Whole Dataframe is going empty. I have tried persisting this dataframe and Also registered to a Temp Table and tried using it. But nothing is working.

Any Help or suggestion would be a big help. Thanks in advance.

-----------------------------solution----------------------------------

So here is how I solved this issue. As Suggested I used MERGE INTO which was a lot faster and as I am using unique transaction ids so I dint have any duplicate issues. Previously I was Updating the table for those transaction IDS then tried to use those same unique IDs with modified values and INSERT INTO the same table.

As a solution I first picked distinct transaction ids from my source and INSERT INTO the table with my updated values , then stored those same list of transaction ids and updated existing older record entries from the table.

val filteringRecordsToExpire = delta.join(collectingSrcSysIdsToExpire, Seq("trans_id"), "leftsemi")
.distinct()

collectingSrcSysIdsToExpire.select(TargetTable.schema.map(f => col(f.name)): _*).write.insertInto(Table)

val sqlUpdateQry =
      s""" MERGE INTO TargetTable as tgtTable
           USING expireSrsIds as source
           ON tgtTable.trans_id = source.trans_id
           AND few more conditions 
           WHEN MATCHED
           THEN UPDATE SET
           expiring older entries

So somehow INSERT then UPDATE works sequentially. But UPDATE then INSERT does not work.

Alex_Ott_18627 · Answer

The foreach is by definition doesn't return any data - you can see from API docs that the return type is Unit. I also don't recommend to update individual records - it will be too slow as it will rewrite the data for each record separately. Instead, use the MERGE operation, with something like this (it's not Scala, just algorithm):

sourceTable
  .as("source")
  .merge(
    dfUpdates.as("updates"),
    "source.id = updates.id")
  .whenMatched
  .updateExpr(
    Map(
      "status" -> "'expired"
    ))

See MERGE documentation for full details. Also instead of updating records, you can delete them.

Using Update on Delta table is changing the state of an intermediate DataFrame

1 Answers