Insert the new,update the changed.

A Copy activity can write straight into a table, but then every rule about what counts as new or changed lives inside the copy's settings. Copying into a staging table first keeps the copy simple and puts those rules in SQL, where you can read, test and change them.

#Two tables, one shape

The target keeps the real data and has a key. The staging table has the same columns and no key: it holds whatever the last copy brought, duplicates and all.

CREATE SCHEMA stg;
GO

CREATE TABLE dbo.Customer (
    CustomerID    int            NOT NULL PRIMARY KEY,
    FirstName     nvarchar(50)   NOT NULL,
    LastName      nvarchar(50)   NOT NULL,
    EmailAddress  nvarchar(50)   NULL,
    CompanyName   nvarchar(128)  NULL,
    ModifiedDate  datetime2      NOT NULL
);

CREATE TABLE stg.Customer (
    CustomerID    int            NULL,
    FirstName     nvarchar(50)   NULL,
    LastName      nvarchar(50)   NULL,
    EmailAddress  nvarchar(50)   NULL,
    CompanyName   nvarchar(128)  NULL,
    ModifiedDate  datetime2      NULL
);

Every staging column allows empty values on purpose. A bad row should arrive and be dealt with in the merge, not stop the copy halfway.

#The copy

In Data Factory, make a dataset for stg.Customer on the Azure SQL Database linked service. Then add a Copy activity to a new pipeline:

• Source: the Customer files in the lake, for example the Parquet folder the first course wrote.
• Sink: the stg.Customer dataset, with Write behavior set to Insert.
• Pre-copy script: TRUNCATE TABLE stg.Customer;

The pre-copy script runs before the rows are written, so staging only ever holds this run's rows. Name the activity Copy customers; the next lessons refer to it by that name.

#Map the columns

Open the Mapping tab and press Import schemas. The names match, so each column maps to itself. Check the types: a date read as text from a CSV file must land as datetime2, or the copy will fail on the first row it cannot convert.

#Try it

Run the pipeline in debug mode, then look at staging:

SELECT COUNT(*) AS Rows, COUNT(DISTINCT CustomerID) AS Customers
FROM stg.Customer;

If the two numbers differ, the source sent some customers twice. That is normal, and the merge in the next lesson keeps only the newest of each.

With this run's rows in staging, one statement can bring the target up to date: update the customers that changed, insert the ones that are new, and leave the rest alone. That statement is MERGE, and it belongs in a stored procedure so the pipeline only has to call it.

#Keep one row per key

MERGE stops with an error if two staging rows match the same target row. So the procedure first keeps the newest row of each customer:

WITH newest AS (
    SELECT *, ROW_NUMBER() OVER (PARTITION BY CustomerID ORDER BY ModifiedDate DESC) AS n
    FROM stg.Customer
    WHERE CustomerID IS NOT NULL
)
SELECT * FROM newest WHERE n = 1;

Rows without a key are left out here. Count them in the run log of the next lesson if they matter to you.

#The procedure

CREATE OR ALTER PROCEDURE etl.MergeCustomer
AS
BEGIN
    SET NOCOUNT ON;
    DECLARE @changes TABLE (Change nvarchar(10));

    WITH newest AS (
        SELECT *, ROW_NUMBER() OVER (PARTITION BY CustomerID ORDER BY ModifiedDate DESC) AS n
        FROM stg.Customer
        WHERE CustomerID IS NOT NULL
    )
    MERGE dbo.Customer AS t
    USING (SELECT * FROM newest WHERE n = 1) AS s
        ON t.CustomerID = s.CustomerID
    WHEN MATCHED AND s.ModifiedDate > t.ModifiedDate THEN
        UPDATE SET FirstName = s.FirstName, LastName = s.LastName, EmailAddress = s.EmailAddress,
                   CompanyName = s.CompanyName, ModifiedDate = s.ModifiedDate
    WHEN NOT MATCHED BY TARGET THEN
        INSERT (CustomerID, FirstName, LastName, EmailAddress, CompanyName, ModifiedDate)
        VALUES (s.CustomerID, s.FirstName, s.LastName, s.EmailAddress, s.CompanyName, s.ModifiedDate)
    OUTPUT $action INTO @changes;

    SELECT
        SUM(CASE WHEN Change = 'INSERT' THEN 1 ELSE 0 END) AS Inserted,
        SUM(CASE WHEN Change = 'UPDATE' THEN 1 ELSE 0 END) AS Updated
    FROM @changes;
END;

The AND s.ModifiedDate > t.ModifiedDate on the update means a row that has not changed is not written again. OUTPUT $action records what happened to each row, and the last SELECT turns that into two numbers.

#Call it from the pipeline

Add a Lookup activity after Copy customers, connected on success, and name it Merge customers. Point it at the Azure SQL dataset, choose Stored procedure and pick etl.MergeCustomer. Leave First row only on.

A Lookup is used here instead of a Stored procedure activity because it returns the procedure's result. After a run, its output holds the counts:

@activity('Merge customers').output.firstRow.Inserted
@activity('Merge customers').output.firstRow.Updated

#Try it

Run the pipeline twice. The first run inserts every customer; the second inserts and updates nothing, because nothing changed. Change one customer at the source, run again, and the second number is 1.

A pipeline that runs every night with nobody watching needs to leave a note: when it ran, what it moved and, when it failed, why. One table and one procedure are enough, and the pipeline writes to them on success and on failure.

#The log table

CREATE TABLE etl.RunLog (
    RunId       nvarchar(64)    NOT NULL,
    Pipeline    nvarchar(128)   NOT NULL,
    TableName   sysname         NOT NULL,
    Status      nvarchar(20)    NOT NULL,
    RowsCopied  int             NULL,
    Inserted    int             NULL,
    Updated     int             NULL,
    Message     nvarchar(4000)  NULL,
    LoggedAt    datetime2       NOT NULL DEFAULT SYSUTCDATETIME()
);

#The procedure that writes it

CREATE OR ALTER PROCEDURE etl.LogRun
    @RunId nvarchar(64), @Pipeline nvarchar(128), @TableName sysname, @Status nvarchar(20),
    @RowsCopied int = NULL, @Inserted int = NULL, @Updated int = NULL, @Message nvarchar(4000) = NULL
AS
BEGIN
    SET NOCOUNT ON;
    INSERT INTO etl.RunLog (RunId, Pipeline, TableName, Status, RowsCopied, Inserted, Updated, Message)
    VALUES (@RunId, @Pipeline, @TableName, @Status, @RowsCopied, @Inserted, @Updated, @Message);
END;

#On success

Add a Stored procedure activity after Merge customers, connected on success, that calls etl.LogRun with:

RunId       @pipeline().RunId
Pipeline    @pipeline().Pipeline
TableName   Customer
Status      Succeeded
RowsCopied  @activity('Copy customers').output.rowsCopied
Inserted    @activity('Merge customers').output.firstRow.Inserted
Updated     @activity('Merge customers').output.firstRow.Updated

#On failure

Add a second Stored procedure activity that calls the same procedure, connected to Copy customers on failure, with Status set to Failed and:

Message     @activity('Copy customers').error.message

Do the same for Merge customers with its own error message. Now a failed run says which step failed and why, in a table you can query.

#Read it

SELECT TOP (20) LoggedAt, TableName, Status, RowsCopied, Inserted, Updated, Message
FROM etl.RunLog
ORDER BY LoggedAt DESC;

The RunId is the same one the Data Factory monitor shows, so a line in the log leads straight to the run's details there.

#What you built

A staging table the copy fills, a merge that updates the changed rows and inserts the new ones, and a log of every run. Put the three steps inside the ForEach of the first course and every table in the control table gets the same treatment.