Avoid duplicates during parallel job run in Azure Synapse

user_5657598 0
Viewed 921

Need your suggestions in developing code in Azure Synapse.

We have a requirement where our jobs will run in parallel at same time and insert data to the same table. During this insert there are changes that duplicate entries will be inserted to the same table. For Example: If Job A and Job B run at same time both with same values then "not exists" or "not in" will fail to work. In this case I will get duplicates from both the job. Primary key or Unique constraint allows duplicates in Azure synapse. Is there any best way to lock tables during data insert. Like if Job A is running then JOB B should not insert the data to same table. Please pour your suggestions as I am new to this. Note: We use stored Procedure to load the data through ADF V2

Thanks, Nandini

azure-synapse azure-data-factory azure
2 Answers

Duplicates must be handled within jobs before inserting data into Azure Synapse. If the duplicates exists between two jobs, then do it after completion of both jobs. It depends really how you are loading data. You can easily manage by creating a temp table instead of directly loading data to final table. Please make sure the structure of temp table should be same as final table (Distribution, Partition, constraints, nullability of the columns) You can use SQL BCP/INSERT TO/CTAS/CTAS with partition switching with stage table to final table.

If you can share specific scenario, it will be helpful to give suggestions relevant to your use case.

user_32791920

I just got the same case and I solved it with Pipeline Runs - Query By Factory

  1. Use a Until activity before the DataFlow activity that writes the values in the table with this expression @equals(activity('pingPL').output.value[0].runId, pipeline().RunId) as follow:

  1. Into the Until activities put a web activity and a wait time:

a. Web activity body - follow docs:

{
  "lastUpdatedAfter": "@{addminutes(utcnow(), -30)}",
  "lastUpdatedBefore": "@{utcnow()}",
  "filters": [
    {
      "operand": "PipelineName",
      "operator": "Equals",
      "values": [
        "pipeline_name_where_writeInSynapse_is_located"
      ]
    },
    {
      "operand": "Status",
      "operator": "Equals",
      "values": [
        "InProgress"
      ]
    }
  ]
}

b. Wait activity 30 sec or whatever make sense

What is happening is, if you trigger several times the same pipeline in parallel the web activity is going to filter each PL status InProgress. It will look like this:

{
    "value": [
        {
            "id": "...",
            "runId": "52004775-5ef5-493b-8a44-ee3fff6bff7b",
            "debugRunId": null,
            "runGroupId": "52004775-5ef5-493b-8a44-ee3fff6bff7b",
            "pipelineName": "synapse_writting",
            "parameters": {
                "region": "NW",
                "unique_item": "a"
            },
            "invokedBy": {
                "id": "80efce4dbda74636878bc99472978ccf",
                "name": "Manual",
                "invokedByType": "Manual"
            },
            "runStart": "2021-10-13T17:24:01.0210945Z",
            "runEnd": "2021-10-13T17:25:06.9692394Z",
            "durationInMs": 65948,
            "status": "InProgress",
            "message": "",
            "output": null,
            "lastUpdated": "2021-10-13T17:25:06.9704432Z",
            "annotations": [],
            "runDimension": {},
            "isLatest": true
        },
        {
            "id": "...",
            "runId": "cf3f5038-ba10-44c3-b8f5-df8ad4c85819",
            "debugRunId": null,
            "runGroupId": "cf3f5038-ba10-44c3-b8f5-df8ad4c85819",
            "pipelineName": "synapse_writting",
            "parameters": {
                "region": "NW",
                "unique_item": "a"
            },
            "invokedBy": {
                "id": "08205e0eda0b41f6b5a90a8dda06a7f6",
                "name": "Manual",
                "invokedByType": "Manual"
            },
            "runStart": "2021-10-13T17:28:58.219611Z",
            "runEnd": null,
            "durationInMs": null,
            "status": "InProgress",
            "message": "",
            "output": null,
            "lastUpdated": "2021-10-13T17:29:00.9860175Z",
            "annotations": [],
            "runDimension": {},
            "isLatest": true
        }
    ],
    "ADFWebActivityResponseHeaders": {
        "Pragma": "no-cache",
        "Strict-Transport-Security": "max-age=31536000; includeSubDomains",
        "X-Content-Type-Options": "nosniff",
        "x-ms-ratelimit-remaining-subscription-reads": "11999",
        "x-ms-request-id": "188508ef-8897-4c21-8c37-ccdd4adc6d81",
        "x-ms-correlation-request-id": "188508ef-8897-4c21-8c37-ccdd4adc6d81",
        "x-ms-routing-request-id": "WESTUS2:20211013T172902Z:188508ef-8897-4c21-8c37-ccdd4adc6d81",
        "Cache-Control": "no-cache",
        "Date": "Wed, 13 Oct 2021 17:29:02 GMT",
        "Server": "Microsoft-IIS/10.0",
        "X-Powered-By": "ASP.NET",
        "Content-Length": "1492",
        "Content-Type": "application/json; charset=utf-8",
        "Expires": "-1"
    },
    "effectiveIntegrationRuntime": "NCAP-Simple-DataMovement (West US 2)",
    "executionDuration": 0,
    "durationInQueue": {
        "integrationRuntimeQueue": 0
    },
    "billingReference": {
        "activityType": "ExternalActivity",
        "billableDuration": [
            {
                "meterType": "AzureIR",
                "duration": 0.016666666666666666,
                "unit": "Hours"
            }
        ]
    }
}

Then the Until expression will evaluate if the first value[0] has runId == pipeline_runid to stop the until activity and run the dataflow that writes in Synapse. Once the PL ends the status will be Succeeded and the Web activity in another job will get the next value[0] with the status InProgress and continue with the next write. This creates a dependency to the parallel jobs to wait until the dataflow validates and writes in table if need it.

user_88285240
Related