Building this pipeline on Azure Data Factory V2

Question

I am currently trying to set up this pipeline on Azure Data Factory V2 (as you can see in the picture attached). In summary this ERP system will export in a monthly basis this report (CSV file with actual and forecast data) and this will be saved in a blob container. As soon as this file CSV is saved, an event trigger should activate this stored procedure that will - in turn - erase all actual data from my fact table in Azure SQL as this gets replaced every month.

Once actual data is deleted, the pipeline would have subsequently a copy activity that would - in turn - copy the CSV report (actuals + forecast) to same fact table in Azure SQL. Once the copy activity is finished, the HTTP logic APP would delete that new CSV file from the blob container. This workflow would be a recurrent event to be carried out very month.

So far I have been able to run these 3 x activities independently. However, when I join them in the same pipeline, I have had some parameters errors when trying to "publish all". Therefore I am not sure whether I need to have the same parameters for each activity in the pipeline?

The JSON code for my pipeline is the following:

{
    "name": "TM1_pipeline",
    "properties": {
        "activities": [
            {
                "name": "Copy Data1",
                "type": "Copy",
                "dependsOn": [
                    {
                        "activity": "Stored Procedure1",
                        "dependencyConditions": [
                            "Succeeded"
                        ]
                    }
                ],
                "policy": {
                    "timeout": "7.00:00:00",
                    "retry": 0,
                    "retryIntervalInSeconds": 30,
                    "secureOutput": false
                },
                "typeProperties": {
                    "source": {
                        "type": "BlobSource",
                        "recursive": false
                    },
                    "sink": {
                        "type": "SqlSink",
                        "writeBatchSize": 10000
                    },
                    "enableStaging": false,
                    "dataIntegrationUnits": 0
                },
                "inputs": [
                    {
                        "referenceName": "SourceDataset_e7y",
                        "type": "DatasetReference",
                        "parameters": {
                            "copyFolder": {
                                "value": "@pipeline().parameters.sourceFolder",
                                "type": "Expression"
                            },
                            "copyFile": {
                                "value": "@pipeline().parameters.sourceFile",
                                "type": "Expression"
                            }
                        }
                    }
                ],
                "outputs": [
                    {
                        "referenceName": "DestinationDataset_e7y",
                        "type": "DatasetReference"
                    }
                ]
            },
            {
                "name": "Stored Procedure1",
                "type": "SqlServerStoredProcedure",
                "policy": {
                    "timeout": "7.00:00:00",
                    "retry": 0,
                    "retryIntervalInSeconds": 30,
                    "secureOutput": false,
                    "secureInput": false
                },
                "typeProperties": {
                    "storedProcedureName": "[dbo].[test_sp]"
                },
                "linkedServiceName": {
                    "referenceName": "AzureSqlDatabase",
                    "type": "LinkedServiceReference"
                }
            },
            {
                "name": "Web1",
                "type": "WebActivity",
                "dependsOn": [
                    {
                        "activity": "Copy Data1",
                        "dependencyConditions": [
                            "Succeeded"
                        ]
                    }
                ],
                "policy": {
                    "timeout": "7.00:00:00",
                    "retry": 0,
                    "retryIntervalInSeconds": 30,
                    "secureOutput": false,
                    "secureInput": false
                },
                "typeProperties": {
                    "url": "...",
                    "method": "POST",
                    "body": {
                        "value": "@pipeline().parameters.BlobName",
                        "type": "Expression"
                    }
                }
            }
        ],
        "parameters": {
            "sourceFolder": {
                "type": "String",
                "defaultValue": "@pipeline().parameters.sourceFolder"
            },
            "sourceFile": {
                "type": "String",
                "defaultValue": "@pipeline().parameters.sourceFile"
            },
            "BlobName": {
                "type": "String",
                "defaultValue": {
                    "blobname": "source-csv/test.csv"
                }
            }
        }
    },
    "type": "Microsoft.DataFactory/factories/pipelines"
}

What is the detail error message? Could you share the pipeline json code? You could find a code button in the top left corner in the UI. — Fang Liu
Hi @FangLiu I have run some tests this afternoon and managed to fix some of the parameter issues. However, the event trigger that should activate the Storage Procedure (as seen above) for some reason is not being triggered when a new blob is created in the cointainer. Do I need to add any parameters to the Storage Procedure activity as this is the first activity in my pipeline? Thanks! — ERR
Also I have just edited the post with the pipeline json code. Thank you! — ERR
For blob event related, please reference this post. stackoverflow.com/questions/51134475/… — Fang Liu
And you can’t put expression into default value. It won’t be evaluated. You need pass value to these parameters with your trigger. You could edit the trigger in the UI. — Fang Liu

Fang Liu Fang Liu · Accepted Answer · 2018-09-13T10:52:30

Please follow this doc to configure you blob event trigger and pass the right value to your parameters.

Building this pipeline on Azure Data Factory V2

1 Answers