topic Re: Delta table upsert - databricks community in Data Engineering

Delta table upsert - databricks community

weldermartins — Fri, 19 Aug 2022 11:35:10 GMT

Hello guys,

I'm trying to use upsert via delta lake following the documentation, but the command doesn't update or insert newlines.

scenario: my source table is separated in bronze layer and updates or inserts are in silver layer.

from delta.tables import *
 
deltaTableVendas = DeltaTable.forPath(spark, '/user/hive/warehouse/bronze.db/vendas')
deltaTableVendasUpdates = DeltaTable.forPath(spark, '/user/hive/warehouse/silver.db/vendas')                                                   
 
dfUpdates = deltaTableVendasUpdates.toDF()
 
deltaTableVendas.alias('vendas') \
  .merge(
    dfUpdates.alias('updates'),
    'vendas.numero_transacao = updates.numero_transacao'
  ) \
  .whenMatchedUpdate(set =
    {
      "numero_transacao": "updates.numero_transacao",
      "numped": "updates.numped",
      "codcli": "updates.codcli",
      "codprod": "updates.codprod",
      "data_venda": "updates.data_venda",
      "quantidade": "updates.quantidade",
      "valor": "updates.valor"
      
    }
  ) \
  .whenNotMatchedInsert(values =
    {
      "numero_transacao": "updates.numero_transacao",
      "numped": "updates.numped",
      "codcli": "updates.codcli",
      "codprod": "updates.codprod",
      "data_venda": "updates.data_venda",
      "quantidade": "updates.quantidade",
      "valor": "updates.valor"
    }
  ) \
  .execute()

Re: Delta table upsert - databricks community

weldermartins — Fri, 19 Aug 2022 16:35:45 GMT

@Hubert Dudek , @Werner Stinckens

Do you have any idea what's going on in this case?

Re: Delta table upsert - databricks community

-werners- — Mon, 22 Aug 2022 06:13:26 GMT

is that the actual location of the data in the delta lake table? seems like a weird place.

The forPath parameter expects the storage location where the data is stored. You point to the location where hive stores its metadata.

Normally this is something like "/mnt/datalake/bronze/deltatable" or something.

Re: Delta table upsert - databricks community

weldermartins — Mon, 22 Aug 2022 12:41:26 GMT

hello @Werner Stinckens , even changing the table to Datalake still doesn't update. In the images as shown the update does not reflect in the datalake.

Re: Delta table upsert - databricks community

-werners- — Mon, 22 Aug 2022 12:55:11 GMT

Right now I am kinda confused what you try to do.

In your code, you are merging the bronze table with data from the silver table (so the bronze table is updated), but you post screenshots of the silver table, which does not change.

And that is normal because the data is updated in the bronze table in your code.

Re: Delta table upsert - databricks community

weldermartins — Mon, 22 Aug 2022 13:56:21 GMT

@Werner Stinckens In fact, the information is inserted in the bronze table, in the silver layer it would receive the update or insertion of the bronze table.

Re: Delta table upsert - databricks community

-werners- — Mon, 22 Aug 2022 14:01:01 GMT

ok so if the bronze table contains the new updated data which you want to propagate to the silver table, you should create a df with the updates from the bronze table (you use the silver table for that)

Next do a merge into silver table using updates (bronze).

That should work.

Re: Delta table upsert - databricks community

weldermartins — Mon, 22 Aug 2022 14:08:57 GMT

@Werner Stinckens But the upsert is already setting the two tables, I don't understand why I should have to set it again.

Re: Delta table upsert - databricks community

weldermartins — Mon, 22 Aug 2022 18:55:40 GMT

I managed to find the solution. In insert and update I was setting the target.

tanks @Werner Stinckens !

delta_df = DeltaTable.forPath(spark, 'dbfs:/mnt/silver/vendas/')                                                   
 
 
delta_df.alias('target').merge(
    source = bronzedf.alias("source"),
    condition = 'target.numero_transacao = source.numero_transacao'
  ) \
  .whenMatchedUpdate(set =
    {
      "numero_transacao": "source.numero_transacao",
      "numped": "source.numped",
      "codcli": "source.codcli",
      "codprod": "source.codprod",
      "data_venda": "source.data_venda",
      "quantidade": "source.quantidade",
      "valor": "source.valor"
      
    }
  ) \
  .whenNotMatchedInsert(values =
    {
      "numero_transacao": "source.numero_transacao",
      "numped": "source.numped",
      "codcli": "source.codcli",
      "codprod": "source.codprod",
      "data_venda": "source.data_venda",
      "quantidade": "source.quantidade",
      "valor": "source.valor"
    }
  ) \
  .execute()

Re: Delta table upsert - databricks community

-werners- — Tue, 23 Aug 2022 07:29:31 GMT

That is what I was trying to explain. Nice you got it working!