Azure Data Factory activity copy: Evaluate column in sink table with @pipeline().TriggerTime

Question

With Data Factory V2 I'm trying to implement a stream of data copy from one Azure SQL database to another.

I have mapped all the columns of the source table with the sink table but in the sink table I have an empty column where I would like to enter the pipeline run time.

Does anyone know how to fill this column in the sink table without it being present in the source table?

Below there is the code of my copy pipeline

{
"name": "FLD_Item_base",
"properties": {
    "activities": [
        {
            "name": "Copy_Team",
            "description": "copytable",
            "type": "Copy",
            "policy": {
                "timeout": "7.00:00:00",
                "retry": 0,
                "retryIntervalInSeconds": 30,
                "secureOutput": false,
                "secureInput": false
            },
            "typeProperties": {
                "source": {
                    "type": "SqlSource"
                },
                "sink": {
                    "type": "SqlSink",
                    "writeBatchSize": 10000,
                    "preCopyScript": "TRUNCATE TABLE Team_new"
                },
                "enableStaging": false,
                "dataIntegrationUnits": 0,
                "translator": {
                    "type": "TabularTranslator",
                    "columnMappings": {
                        "Code": "Code",
                        "Name": "Name"
                    }
                }
            },
            "inputs": [
                {
                    "referenceName": "Team",
                    "type": "DatasetReference"
                }
            ],
            "outputs": [
                {
                    "referenceName": "Team_new",
                    "type": "DatasetReference"
                  }
              ]
          }
      ]
  }
}

In my sink table I already have the column data_loadwhere I would like to insert the pipeline execution date, but I did not currently map it.

Jay Gong Jay Gong · Accepted Answer · 2018-10-01T06:32:55

Based on your situation, please configure SQL Server stored procedure in your SQL Server sink as a workaround.

Please follow the steps from this doc:

Step 1: Configure your Sink dataset:

Step 2: Configure Sink section in copy activity as follows:

Step 3: In your database, define the table type with the same name as sqlWriterTableType. Notice that the schema of the table type should be same as the schema returned by your input data.

    CREATE TYPE [dbo].[testType] AS TABLE(
    [ID] [varchar](256) NOT NULL,
    [EXECUTE_TIME] [datetime] NOT NULL
)
GO

Step 4: In your database, define the stored procedure with the same name as SqlWriterStoredProcedureName. It handles input data from your specified source, and merge into the output table. Notice that the parameter name of the stored procedure should be the same as the "tableName" defined in dataset.

Create PROCEDURE convertCsv @ctest [dbo].[testType] READONLY
AS
BEGIN
  MERGE [dbo].[adf] AS target
  USING @ctest AS source
  ON (1=1)
  WHEN NOT MATCHED THEN
      INSERT (id,executeTime)
      VALUES (source.ID,GETDATE());
END

Azure Data Factory activity copy: Evaluate column in sink table with @pipeline().TriggerTime

2 Answers