Module 01A - Incremental Copy to Raw (using Change Data Capture)

< Previous Module - Home - Next Module >

⏱️ Estimated Duration

30 minutes

🤔 Prerequisites

Lab environment deployed

📢 Introduction

In this module, we will setup a Synapse Pipeline to incrementally copy data from an OLTP source (Azure SQL Database) to a Data Lake (Azure Data Lake Storage Gen2), leveraging Change Data Capture technology to isolate changes.

The pipeline will perform an initial check to see if any changes (new records or modifications to existing records) exist in the source system since the last load.
If there are changes, the new data will be copied to the raw layer of the Data Lake.

flowchart LR
ds1[(Azure SQL DB\n CDC enabled)]
ds2[(Data Lake\nraw)]
ds1-.changeCount.->a1
ds1-.source\ncdc.dbo_Customers_CT.->a3
a3-."sink\n01-raw/wwi/customers/$fileName.csv".->ds2

subgraph p["Pipeline (C1 - pipelineIncrementalCopyCDC)"]
a1[Lookup\nGetChangeCount]
a1-->a2
    subgraph a2[If Condition\nHasChangedRows]
    a3[Copy data\ncopyIncrementalData]
    end
end

🎯 Objectives

Enable Change Data Capture on source table(s)
Create a Pipeline to copy data changes to the data lake

Source Environment (dbo.Customers)
Linked Service (Azure SQL Database)
Integration Dataset (Azure SQL Database - Table)
Integration Dataset (Azure Data Lake Storage Gen2 - Raw)
Pipeline (Lookup)
Pipeline (If Condition, Copy data)
Load Additional Data into dbo.Customers
Rerun Pipeline to Copy Additional Data

1. Source Environment (dbo.Customers)

Initialize the source environment by creating a table, enabling CDC on the table, and populating the table with data.

Navigate to the SQL database
Click Query editor
Copy and paste your Login and Password from the code snippets below

Login
```
sqladmin
```
Password
```
sqlPassword!
```

To create the source table, copy and paste the code snippet below and click Run

CREATE TABLE Customers (
    CustomerID int IDENTITY(1,1) PRIMARY KEY,
    CustomerAddress varchar(255) NOT NULL
);

To enable change data capture on the source table, copy and paste the code snippet below and click Run

EXEC sys.sp_cdc_enable_db;
EXEC sys.sp_cdc_enable_table  
    @source_schema = N'dbo',  
    @source_name   = N'Customers',  
    @role_name     = NULL,
    @supports_net_changes = 1;

To load the source table with data, copy and paste the code snippet below and click Run

INSERT INTO dbo.Customers (CustomerAddress)
VALUES
    ('82 Margate Drive, Sheffield S4 8FQ'),
    ('135 High Barns, Ely, CB7 4RH'),
    ('39 Queen Annes Drive, Bedale, DL8 2EL');

↥ back to top

2. Linked Service (Azure SQL Database)

Creating a linked service provides Azure Synapse Analytics the necessary information to establish connectivity to an external resource, in this case, an Azure SQL Database.

Navigate to the Synapse workspace
Open Synapse Studio
Navigate to the Manage hub
Click Linked services
Click New
Search SQL, select Azure SQL Database, and click Continue
Rename the Linked Service to AzureSqlDatabase
Select the target Azure SQL Database by selecting the Azure subscription, Server name and Database name
Set the Authentication type to SQL authentication
Copy and paste the User name
```
sqladmin
```
Copy and paste the Password
```
sqlPassword!
```
Click Test connection
Click Create

↥ back to top

3. Integration Dataset (Azure SQL Database - Table)

An integration dataset is simply a named reference to data that can be used in an activity as an input or output. In this example, we are creating a reference to tables within our Azure SQL Database and leveraging parameters to be able to dynamically specify the schema and table name at runtime.

Navigate to the Data hub
Switch to the Linked tab
Click the [+] icon to add a new resource and click Integration dataset
Search SQL, select Azure SQL Database, and click Continue
Rename the Integration Dataset to AzureSqlTable
Select the Linked service AzureSqlDatabase
Click OK
Switch the the Parameters tab
Click New
Set the Name to schema
Click New
Set the Name to table
Switch to the Connection tab
Beneath the Table dropdown menu, select the Edit checkbox
Click inside the first text input for Table and click Add dynamic content
Under Parameters, select schema and click OK
Click inside the second text input for Table and click Add dynamic content
Under Parameters, select table and click OK
Click Publish all
Click Publish

↥ back to top

4. Integration Dataset (Azure Data Lake Storage Gen2 - Raw)

In this example, we are creating a reference to delimited text files (i.e. CSV) within our Azure Data Lake Gen2 Storage Account and leveraging parameters to be able to dynamically specify the folder path and file name at runtime.

Navigate to the Data hub
Switch to the Linked tab
Click the [+] icon to add a new resource and click Integration dataset
Search Data Lake, select Azure Data Lake Storage Gen2, and click Continue
Select DelimitedText and click Continue
Rename the integration dataset to AdlsRawDelimitedText
Select the Azure Synapse Analytics workspace default storage Linked service
Click the browse icon
Select 01-raw and click OK
Select First row as header and click OK
Switch to the Parameters tab
Click New
Set the Name to folderPath
Click New
Set the Name to fileName
Switch to the Connection tab
Click inside the Directory text input and click Add dynamic content
Under Parameters, select folderPath and click OK
Click inside the File text input and click Add dynamic content
Under Parameters, select fileName and click OK
Click Publish all
Click Publish

↥ back to top

5. Pipeline (Lookup)

A pipeline is a data-driven workflow, logically grouping activities to perform a task (e.g. ingest and load). Once our pipeline is created, we will add our first activity - Lookup. The Lookup activity can retrieve a dataset from any of the data sources supported by Synapse pipelines. In this example, we will be executing SQL against our Azure SQL Database to determine the number of changes that have occurred to the target table for a given time period.

Navigate to the Integrate hub
Click the [+] icon to add a new resource and click Pipeline
Rename the pipeline to C1 - pipelineIncrementalCopyCDC
Under Parameters click New
Set the Name to triggerStartTime
Click New
Set the Name to triggerEndTime
Within Activities, search for Lookup, and drag the Lookup activity onto the canvas
Rename the activity GetChangeCount
Switch to the Settings tab
Set the Source dataset to AzureSqlTable
Set the Dataset property schema to cdc
Set the Dataset property table to dbo_Customers_CT
Set the Use query property to Query
Click inside the Query text input and click Add dynamic content
Copy and paste the code snippet and click OK
```
@concat('DECLARE @begin_time datetime, @end_time datetime, @from_lsn binary(10), @to_lsn binary(10); 
SET @begin_time = ''',pipeline().parameters.triggerStartTime,''';
SET @end_time = ''',pipeline().parameters.triggerEndTime,''';
SET @from_lsn = sys.fn_cdc_map_time_to_lsn(''smallest greater than or equal'', @begin_time);
SET @to_lsn = sys.fn_cdc_map_time_to_lsn(''largest less than'', @end_time);
IF (@from_lsn IS NOT NULL AND @to_lsn IS NOT NULL AND @from_lsn < @to_lsn)
SELECT count(1) changecount FROM cdc.fn_cdc_get_net_changes_dbo_Customers(@from_lsn, @to_lsn, ''all'')
ELSE SELECT 0 changecount')
```
🤔 What does the dynamic content do?

At runtime, the pipeline will pass parameters triggerStartTime and triggerEndTime to the @concat function which will result in a SQL statement.

The query performs the following high-level steps:
- DECLARE variables (@begin_time, @end_time, @from_lsn, and @to_lsn)
- SET the variable values
- Calculates the number of net changes within the given time period
The query is able to achieve this by leveraging CDC functions such as:
- sys.fn_cdc_map_time_to_lsn which returns a log sequence number (LSN) for a given datetime
- cdc.fn_cdc_get_net_changes_<capture_instance> which returns the net changes for a specified LSN range.
Click Preview data
Provide a value for triggerStartTime that is a date before today (e.g. 2022-01-01)
Provide a value for triggerEndTime that is a data in the future (e.g. 9999-12-31)
Click OK
You should see a changecount of 3, close the Preview data window
On the Integrate pane, click the ellipses button next to Pipelines, and select New folder
Copy and paste the Folder name from the snippet below and click Create
```
Customers
```
Click on the ellipses button next to C1 - pipelineIncrementalCopyCDC and select Move item
Select the Customers folder and click Move
Click Publish all
Click Publish

↥ back to top

6. Pipeline (If Condition, Copy data)

In this step, we will be adding an If Condition activity to our pipeline. The If Condition activity provides comparable functionality to an if statement found in programming languages. It can execute a set of activities if a condition evaluates to true, and another set of activities if the condition evaluates to false. In this example, we are going to only proceed with a subsequent Copy activity if the number of changes detected is greater than zero.

Within Activities, search for If, and drag the If Condition activity onto the canvas
Click and drag on the green button from the Lookup to the If Condition to establish a connection
Rename the If Condition activity to HasChangedRows
Switch to the Activities tab
Click inside the Expression text input and click Add dynamic content

Copy and paste the code snippet and click OK

@greater(int(activity('GetChangeCount').output.firstRow.changecount),0)

Within the True case, click the pencil icon
Within Activities, search for Copy, and drag the Copy data activity onto the canvas
Rename the Copy activity to copyIncrementalData
Switch to the Source tab
Set Source dataset to AzureSqlTable
Under Dataset properties, set the schema to cdc
Under Dataset properties, set the table to dbo_Customers_CT
Set Use query to Query
Click inside the Query text input and click Add dynamic content

Copy and paste the code snippet and click OK

@concat('DECLARE @begin_time datetime, @end_time datetime, @from_lsn binary(10), @to_lsn binary(10); 
SET @begin_time = ''',pipeline().parameters.triggerStartTime,''';
SET @end_time = ''',pipeline().parameters.triggerEndTime,''';
SET @from_lsn = sys.fn_cdc_map_time_to_lsn(''smallest greater than or equal'', @begin_time);
SET @to_lsn = sys.fn_cdc_map_time_to_lsn(''largest less than'', @end_time);
SELECT CustomerID, CustomerAddress FROM cdc.fn_cdc_get_net_changes_dbo_Customers(@from_lsn, @to_lsn, ''all'')')

Switch to the Sink tab
Set Sink dataset to AdlsRawDelimitedText
Under Dataset properties, set the folderPath to wwi/customers
Under Dataset properties, click inside the fileName text input and click Add dynamic content

Copy and paste the code snippet and click OK

@concat(formatDateTime(pipeline().parameters.triggerStartTime,'yyyyMMddHHmmssfff'),'.csv')

Navigate back up to the pipeline and click Publish all
Click Publish
Click Debug
Provide a value for triggerStartTime that is a date before today (e.g. 2022-01-01)
Provide a value for triggerEndTime that is a data in the future (e.g. 9999-12-31)
Click OK
When the pipeline run is complete, under the Output tab, click the Details icon of the Copy data activity to confirm that three rows have been written to the data lake.
You can also navigate to the Data hub, browse the data lake folder structure under the Linked tab to 01-raw/wwi/customers, right-click the CSV file and select New SQL Script > Select TOP 100 rows
Modify the SQL statement to include HEADER_ROW = TRUE within the OPENROWSET function and click Run

↥ back to top

7. Load Additional Data into dbo.Customers

Before we can test that our pipeline is able to successfully isolate and copy changes from a particular time period, we must perform changes to our source table (e.g. UPDATE existing rows, INSERT new rows).

Navigate to the SQL database
Click Query editor
Copy and paste your Login and Password from the code snippets below

Login
```
sqladmin
```
Password
```
sqlPassword!
```

Copy and paste the code snippets below and click Run

UPDATE dbo.Customers SET CustomerAddress = 'Guyzance Cottage, Guyzance NE65 9AF' WHERE CustomerID = 3;
INSERT INTO dbo.Customers (CustomerAddress)
VALUES
    ('322 Fernhill, Mountain Ash, CF45 3EN'),
    ('381 Southborough Lane, Bromley, BR2 8BQ');
SELECT * FROM [dbo].[Customers];

Copy and paste the code snippet below and click Run. Note: There may be some latency between the changes being executed and the changes being recorded in the related CDC table. You may need to wait a minute or two between steps to get the correct start_time and end_time values.

DECLARE @max_lsn binary(10);
SET @max_lsn = sys.fn_cdc_get_max_lsn();  
SELECT
CONVERT(varchar(16), DATEADD(minute, -1, sys.fn_cdc_map_lsn_to_time(@max_lsn)), 20) as start_time,
CONVERT(varchar(16), DATEADD(minute, 1, sys.fn_cdc_map_lsn_to_time(@max_lsn)), 20) as end_time

Copy and paste the start_time and end_time values into a text editor (e.g. Notepad). This will be used as input for the pipeline rerun to isolate the second batch of changes made to the dbo.Customers table.

↥ back to top

8. Rerun Pipeline to Copy Additional Data

Using the start_time and end_time values from the previous step, we will rerun our pipeline and confirm that the changes have been copied to the Azure Data Lake Gen2 Storage Account.

Navigate to the Synapse workspace
Open Synapse Studio
Navigate to the Integration hub
Open pipeline C1 - pipelineIncrementalCopyCDC
Click Debug
Copy and paste the start_time and end_time values into the triggerStartTime and triggerEndTime parameters and click OK
When the pipeline run is complete, under the Output tab, click the Details icon of the Copy data activity to confirm that three rows have been written to the data lake.
You can also navigate to the Data hub, browse the data lake folder structure under the Linked tab to 01-raw/wwi/customers, right-click the second CSV file and select New SQL Script > Select TOP 100 rows
Modify the SQL statement to include HEADER_ROW = TRUE within the OPENROWSET function and click Run