Azure Synapse Upsert Record into Dedicated Sql Pool

cva 0
Viewed 90

We have requirement that we need to fetch json data from the datalake storage and insert/update data into synapse tables based on the lastmodified field in the source json and table column.

we need to perform either insert/update record based on following conditions.

if(sourceJson.id==table.id) //assume record already exists
{
   if (SourceJson.lastmodified > table.lastmodified){

       //update existing record

   }   
   else if(SourceJson.lastmodified<table.lastmodified){
        //ignore record
    }

}

else{
   
    //insert record 
}

is there any way to achieve this, if there please help me on this by sharing any sample flow.

Thanks

azure-data-factory azure-synapse
1 Answers

The copy data activity and azure dataflows both have an option to Upsert. But they would not help your requirement.

  • Since you have a key column id and also have a special condition based on which you want to either insert or ignore a record, you can create a stored procedure first in your azure synapse dedicated pool.

  • The following is the data available in my table: enter image description here

  • The following is the data available in my JSON:

[
{
"id":1,
"first_name":"Ana",
"lastmodified":"2022-09-10 07:00:00"
},
{
"id":2,
"first_name":"Cassy",
"lastmodified":"2022-09-07 07:00:00"
},
{
"id":5,
"first_name":"Topson",
"lastmodified":"2022-09-10 07:00:00"
}
]
  • Use lookup to read the input JSON file. Create a dataset, uncheck first row only and run it. The following is my debug output:

enter image description here

  • Now, create a stored procedure. If I have created it directly in my Synapse pool (You can use use script activity to create it).
CREATE  PROCEDURE mymerge
@array varchar(max)
AS
BEGIN
--inserting records whose id are not present in table
insert  into demo1 SELECT * FROM OPENJSON(@array) WITH ( id int,first_name varchar(30),lastmodified datetime) where id not  in (select id from demo1);

--using MERGE to update records based on matching id and lastmodified column condition
MERGE  into demo1 as tgt
USING (SELECT * FROM OPENJSON(@array) WITH ( id int,first_name varchar(30),lastmodified datetime) where id in (select id from demo1)) as ip
ON (tgt.id = ip.id and ip.lastmodified>tgt.lastmodified)
WHEN  MATCHED  THE
UPDATE  SET tgt.first_name = ip.first_name, tgt.lastmodified = ip.lastmodified;
END
  • Create a stored procedure activity. Select the above created Stored procedure and pass the lookup output array as a string parameter to stored procedure to get the required result.
@string(activity('Lookup1').output.value)

enter image description here

  • Running this would give the required result.

enter image description here

Saideep Arikontham0
Related