How to do arithmetic operations when doing delta table updates?

user_7418045 0
Viewed 26

I have a delta table old, I want to merge it with new. In the new table, there are some id values which are also present in old table. I want to update cons values for the overlapping ids by summing up the old and new table cons values. How to do that?

delta-lake pyspark delta
1 Answers

Try this:

In delta table updates, you can do arithmetic operations just like what you would have done when creating any new spark column.

import pyspark.sql.functions as F
from delta.tables import *

spark.createDataFrame([{"id":i, "cons":1, "cons2":1} for i in range(500)])\
.write.format("delta").mode("overwrite").option("overwriteSchema", "true")\
.save("dbfs:/FileStore/anmol/sample_events_croma_before")

new = spark.createDataFrame([{"id":i, "cons":1, "cons2":1} for i in range(450, 550)])

old = DeltaTable.forPath(spark, "dbfs:/FileStore/anmol/sample_events_croma_before")

old.alias('old')\
.merge(source=new.alias('new')\
       , condition="old.id = new.id")\
.whenMatchedUpdate(set={
  "id": "new.id",
  "cons": "old.cons + new.cons",
  "cons2": F.col("old.cons2") + F.col("new.cons2"),
})\
.whenNotMatchedInsert(values={
  "id": "new.id",
  "cons": "new.cons",
})\
.execute()

You can also pass in complex boolean operations on columns in the condition parameter of this method.

For example, instead of old.id=new.id, we can pass

condition=((F.col("old.id")==F.col("new.id"))&(F.col("new.id")<F.lit(10))
user_74180450
Related