Implement Change Data Streams with Azure SQL Database, Change Tracking, and Azure SQL Bindings

Brian Spendolini

Building on my first Azure SQL Bindings blog post and taking inspiration from Drew Skwiers-Koballa’s post on SQL Bindings, we are going to create a change data stream with Azure SQL Database. Now, usually, the first part of these blogs is a prerequisite section that goes over what you need to install locally to get started. From talking to many of you in person and getting questions/emails on this process, this post is going to use GitHub codespaces; nothing to install locally or configure. I’ve created a project here that you can fork and create a codespace to get you up and running in minutes.

Starting the codespace

To start a codespace, just fork the project located here. Then, click the green Code button on the forked repository. Click the Codespaces tab followed by clicking the green Create codespace on main button.

This will create the codespace using the devcontainer.json file in the .devcontainer directory. Once the codespace is up and running, it will then run the install-dev-tools.sh file in the scripts directory to configure the codespace with all the necessary tools for this blog post and for developing with the Azure SQL Database in general.

Change data stream components

For this project, we are going to use the following services/features locally in the codespace and in Azure.

• SQL Server 2022/Azure SQL Database: This will be created using docker in docker within the codespace.
• Azure Functions. Azure Functions will act as the bridge between the database and the Azure Event Hub.
• SQL Trigger Bindings will be watching a table in the database for changes (via Change Tracking).
• Azure Event Hub  will recieve the data change onces the trigger files and the function passes it over. There we will capture the change event and place it in Azure Blob Storage.

Create a table with change tracking

Once, the codespace has started up, go to the terminal and issue the following commands. We are going to create a directory for our database objects and a SQL Database Project:

mkdir database
cd database
dotnet new sqlproj -n "devDB" -tp "SqlAzureV12"

sqlcmd config connection-strings

Create a table in SQL Database Projects

CREATE TABLE [dbo].[customer] (
    [customer_id]        INT            IDENTITY (1, 1) NOT NULL PRIMARY KEY CLUSTERED ([customer_id] ASC),
    [customer_name]      NVARCHAR (200) NOT NULL,
    [customer_email]     NVARCHAR (200) NOT NULL,
    [customer_address]   NVARCHAR (500) NOT NULL
);

Viewing the published table

Enable change tracking

ALTER DATABASE [devDB]
SET CHANGE_TRACKING = ON;
GO

ALTER TABLE [dbo].[customer] ENABLE CHANGE_TRACKING;
GO

Create an Azure Function

Back to the terminal

---

You may have to click the terminal tab on the window bar on the bottom of the page. When you deployed the SQL Database Project, it may have moved from the terminal window to the Output window.

cd /workspaces/codespace-for-DB-Devs

func init

Adding necessary libraries to the project

dotnet add package Microsoft.Azure.WebJobs.Extensions.Sql --prerelease

dotnet add package Azure.Messaging.EventHubs

dotnet add package Microsoft.Azure.WebJobs.Extensions.EventHubs

namespace Company.Function;
public class Customer
{
    public int customer_id { get; set; }
    public string customer_name { get; set; }
    public string customer_email { get; set; }
    public string customer_address { get; set; }
}

Create the SQL trigger function

Adding the SQL Bindings code

using System.Collections.Generic;
using Microsoft.Azure.WebJobs;
using Microsoft.Extensions.Logging;
using Microsoft.Azure.WebJobs.Extensions.Sql;
using System.Threading.Tasks;
using System.Text.Json;
namespace Company.Function;
public static class streamCustomers
{
    [FunctionName("changeDataStream")]
    public static async Task RunAsync(
        [SqlTrigger("[dbo].[customer]", "SqlConnectionString")]
            IReadOnlyList<SqlChange<Customer>> changes,
        ILogger logger)
   {
      foreach (SqlChange<Customer> change in changes)
      {
          var customer = JsonSerializer.Serialize(change.Item);
          var message = $"{change.Operation} {customer}";
          logger.LogInformation(message);
        }
    }
}

"SqlConnectionString": "Server=localhost,1433;Database=devDB;User ID=vscode;Password=XXXXX;Encrypt=True;TrustServerCertificate=True;Connection Timeout=30;",

and be sure to replace the XXXXX with the password of the database that was created before. If you forgot the password, run the following in the terminal to find it again:

sqlcmd config connection-strings

func host start

insert into dbo.customer values(N'Bill', N'bill@computer.commm', N'Anytown, Anycity 12345');

[2023-05-11T19:18:03.124Z] Executing 'changeDataStream' (Reason='New change detected on table '[dbo].[customer]' at 2023-05-11T19:18:03.0902305Z.', Id=541ff09e-54ac-48e8-8d17-bbcc9a451432)
[2023-05-11T19:18:03.176Z] Insert{"customer_id":1,"customer_name":"Bill","customer_email":"bill@computer.commm","customer_address":"Anytown, Anycity 12345"}
[2023-05-11T19:18:03.195Z] Executed 'changeDataStream' (Succeeded, Id=541ff09e-54ac-48e8-8d17-bbcc9a451432, Duration=93ms)

using System.Collections.Generic;
using Microsoft.Azure.WebJobs;
using Microsoft.Extensions.Logging;
using Microsoft.Azure.WebJobs.Extensions.Sql;
using System.Threading.Tasks;
using Azure.Messaging.EventHubs;
using System.Text.Json;
namespace Company.Function;
public static class streamCustomers
{
    [FunctionName("changeDataStream")]
    public static async Task Run(
        [SqlTrigger("[dbo].[customer]", "SqlConnectionString")]
            IReadOnlyList<SqlChange<Customer>> changes,
        [EventHub("HUB_NAME", Connection = "EventHubConnectionAppSetting")]
            IAsyncCollector<EventData> outputEvents,
        ILogger logger)
    {
        foreach (SqlChange<Customer> change in changes)
        {
            var customer = JsonSerializer.Serialize(change.Item);
            var message = $"{change.Operation} {customer}";
            logger.LogInformation(message);
            await outputEvents.AddAsync(new EventData(message));
        }
    }
}

"EventHubConnectionAppSetting": "XXXXX",

func host start

insert into dbo.customer values(N'Ellie', N'ellie@computer.commm', N'Anytown, Anycity 12345');

[2023-05-12T21:47:03.447Z] Executing 'changeDataStream' (Reason='New change detected on table '[dbo].[customer]' at 2023-05-12T21:47:03.4465111Z.', Id=5c7077d5-cec9-424a-bc51-17f08b8677bb)
[2023-05-12T21:47:03.448Z] Insert {"customer_id":2,"customer_name":"Ellie","customer_email":"ellie@computer.commm","customer_address":"Anytown, Anycity 12345"}
[2023-05-12T21:47:03.561Z] Executed 'changeDataStream' (Succeeded, Id=5c7077d5-cec9-424a-bc51-17f08b8677bb, Duration=114ms)

Passwordless Authentication

Server=productiondatabase.database.windows.net; Authentication=Active Directory Managed Identity; Database=prodDB

Outbox pattern with stored procedures

Brian Spendolini Senior Product Manager, Azure SQL Database

Follow