DEV Community: Tasio Guevara

Unlock innovation using data from your on-premises databases (III) - Event-driven apps

Tasio Guevara — Mon, 09 May 2022 16:09:46 +0000

In my previous blog post, I described how to capture data changes in a relational database and push them as records in a Kinesis data stream.

In this new post, I will describe how to convert those database operations into business events of potential interest in near real-time, so other applications can react to them.

With the solution deployed in the previous post, now I have at my disposal a steady stream of all the changes that are committed to tables of interest in the source database. Now, it's a matter of consuming those changes and turning them into a more actionable construct: business events.

For that, I am going to use two AWS Lambda functions and the Lambda capability to filter event sources from Amazon Kinesis Data Streams. This filtering will help me reduce the number of invocations, reducing cost and potentially allowing to scale my application more easily.

When the Lambda functions receive new change data capture (CDC) records, they will, based on some business logic, publish events into the default Amazon EventBridge event bus. The idea behind these functions is to convert database operations into "business events" that carry more business context in their definition and data. Once those events are published in EventBridge, you can use the rich variety of supported EventBridge targets to build event-driven applications that react to them.

For this example, I will publish two business events:

NegativeProductReviewReceived, issued when a product receives a bad review (less than 4 stars rating).
ProductStockOverbooked, issued When a transaction is filed with a quantity of a product above the available quantity in stock. Note that this is just to illustrate the use case, depending on the real business logic, a system might not allow an order that has quantities above existing stock.

As aforementioned, I'll use two separate Lambda functions for processing records from the two source tables involved. I could use the same function and two filters, but I think that architecturally it makes more sense this way, because it's likely that two separate teams would take care of those business areas.

The next sections describe the logic for each function.

Lambda function that processes changes on the ProductReview table

This function will publish the NegativeProductReviewReceived event. It will be triggered every time there is an update or insert operation on the Production.ProductReview table and there is a review with a rating below 4. Although the event filter will be configured to check for those specific conditions, I will also include that logic in the function code. This way, if I want to use this function to generate other events, I can simply loosen up the triggering conditions and add that new logic in the function code. Note that, with that logic in the function, I could already configure a more permissive filter to trigger the function, but this way I avoid unnecessary function invocations and showcase how powerful the Lambda event filtering feature is.

Some additional logic in the function will handle the special case of an update; the function will check that the previous value was not lower than or equal to the new value, avoiding issuing an event related to the same review twice.

Lambda function that processes changes on the TransactionHistory table

This function will publish the ProductStockOverbooked event. It will be triggered every time there is an update or insert operation in the Production.TransactionHistory table. It will query the original database to check the existing stock of the product in the transaction and, if the amount is lower than the one found in the transaction, it will issue the ProductStockOverbooked event.

The architecture

This is the architecture for this solution. The most notable characteristic is that the Lambda function that processes the TransactionHistory table is configured to access resources in the same VPC where the AWS DMS replication instance sits. This is to provide connectivity to the source database and enable the function to query data to fulfill its business logic. Due to this and the function needing to access EventBridge to publish events, I will also need to deploy a VPC Endpoint for that service.

Deploying the solution

To make it easier to package and deploy the Lambda functions and all the necessary resources (e.g.: IAM roles for the functions), I will use a very convenient tool when working with serverless applications: the AWS Serverless Application Model (SAM). SAM uses templates similar to those of AWS CloudFormation, but it takes care of downloading and packaging all your function's dependencies into a .zip file or a Docker container, depending on the package type you choose. Furthermore, if you have runtime-specific dependencies (e.g.: Python packages that use C/C++ extensions), you can easily install a Labmda-compatible package by using Docker locally and the --use-container flag when building the Lambda package. Finally, it uploads the function package or container to Amazon S3 or Amazon ECR, respectively, so you can deploy them using CloudFormation.

If you want to follow the next steps, you need to install the AWS SAM CLI. SAM provides project templates to initialize different kinds of sample serverless applications. I will skip that part, but I encourage you to check the built-in project templates or create your own using Cookiecutter.

Set up the following project directory and file structure under a folder of your choice (e.g.: cdc2event)

cdc2event
├── reviews
│   ├── app.py
│   └── requirements.txt
├── transactions
│   ├── app.py
│   └── requirements.txt
└── template.yaml

Edit the app.py file under the reviews directory and paste the following code. You can read details about what the code does in the comments.

import base64
import json
from dateutil.parser import parse
import boto3
import logging

logger = logging.getLogger(name="cdc2eventLogger")
logger.setLevel(level=logging.DEBUG)

event_bridge = boto3.client("events")

def lambda_handler(event, context):
    """
    This function processes CDC records from inserted and updated product reviews that have a rating below 4.
    For each of the records, it will publish an event in Amazon EventBridge.
    If the record has been updated, it will only publish an event if the previous value was above or equal to 4.
    """
    events = [] # This list will containt the events to publish on the EventBridge event bus.

    # An event can contain several Kinesis records.
    # Each Kinesis record contains data encoded in Base64, so it has to be decoded to a string.
    # The replication task was configured to deliver JSON objects, so these have
    # to be parsed into Python objects using the json package.
    for record in event["Records"]:
        payload = base64.b64decode(record["kinesis"]["data"])
        data = json.loads(payload)
        logger.debug(f"Processing data: {data}")
        # If the record comes from a database update operation,
        # check whether the rating value was above or equal to 4.
        if data["metadata"]["operation"] == "update" and int(data["before"]["Rating"]) >= 4 or \
            data["metadata"]["operation"] == "insert" and data["data"]["Rating"] < 4:
            # Create an EventBridge event and add it to the list
            events.append(
                create_event(data, "com.example.cdc2event")
            )
        else:
            logger.debug(f"Ignoring event: {data}")

    # Publish the events in the default EventBridge bus
    logger.debug(f"Publishing to EventBridge: {events}")
    if events:
        event_bridge.put_events(
            Entries=events
        )

    return

def create_event(record, source):
    """
    This function returns an EventBridge event based on the information within a record
    """
    # The detail of the event is a dictionary with any arbitrary structure.
    # In this case it contains all the information from the data and metadata items
    # provided by DMS, as it may be relevant for applications downstream.
    # Depending on the use case, removing certain fields might be needed for regulatory purposes.
    event = {
        "Time": parse(record["metadata"]["timestamp"]),
        "Source": source,
        "DetailType": "NegativeProductReviewReceived",
        "Detail": json.dumps({
            "data": record["data"],
            "metadata": record["metadata"]
        })
    }

    return event

That piece of code uses an external Python package, so you need to include it in the Lambda package. For that, edit the requirements.txt file under the reviews directory and paste the following text.
```
python-dateutil
```

Edit the app.py file under the transactions directory and paste the following code. Check the comments for further insights into it.

import os
import base64
import json
import logging
from dateutil.parser import parse
import boto3
import pymssql

logger = logging.getLogger(name="cdc2eventLogger")
logger.setLevel(level=logging.DEBUG)

eventbridge = boto3.client("events")
secretsmanager = boto3.client("secretsmanager")

connection_details = json.loads(
    secretsmanager.get_secret_value(SecretId=os.environ["DB_SECRET_ARN"])["SecretString"])

database = os.environ["DB_NAME"]

MAX_DB_READ_ATTEMPTS = 3

def connect_to_database():
    global db_connection
    db_connection = None

    try:
        db_connection = pymssql.connect(
            host=connection_details["host"],
            port=connection_details["port"],
            user=connection_details["username"],
            password=connection_details["password"],
            database=database
        )
    except pymssql.InterfaceError as e:
        logger.debug(f"Error connecting to the database, please check the host name.\n {e}")

    except pymssql.DatabaseError as e:
        logger.debug(f"Error connecting to the database, check the credentials. \n {e}")

    return db_connection

# Initialize the connection in the global scope.
# This way, the connection will be reused across function executions.
db_connection = connect_to_database()

def lambda_handler(event, context):
    """
    This function processes CDC records from inserted and updated product reviews that have a rating below 4.
    For each of the records, it will publish an event in Amazon EventBridge.
    If the record has been updated, it will only publish an event if the previous value was above or equal to 4.
    """
    events = [] # This list will containt the events to publish on the EventBridge event bus.

    # An event can contain several Kinesis records.
    # Each Kinesis record contains data encoded in Base64, so it has to be decoded to a string.
    # The replication task was configured to deliver JSON objects, so these have
    # to be parsed into Python objects using the json package.
    records = map(decode_kinesis_record, event["Records"])

    # Filter out the records where the quantity has not changed to avoid duplicate events.
    records = list(filter(
        lambda x: x["data"]["Quantity"] != int(x.get("before", {}).get("Quantity", "-1")),
        records
    ))
    logger.debug(f"Processing changed records: {records}")

    if len(records) == 0:
        logger.debug("No records to process.")
        return
    # Extract all the product identifiers in all the records to be processed.
    # This way, there is only one round-trip to the database
    product_ids = [record["data"]["ProductID"] for record in records]
    stock = get_stock(product_ids)

    # Check that the inventory read was successful
    if stock:
        # Create an event for every order that has a quantity above the aggregated stock in inventory
        for record in records:
            product_id = record["data"]["ProductID"]
            if record["data"]["Quantity"] > stock[str(product_id)]:
                # Add the stock quantity to the business event
                record["data"]["StockQuantity"] = stock[str(product_id)]
                events.append(
                    create_event(record, "com.example.cdc2event")
                )
            else:
                logger.debug(f"Ignoring event: {record}")

    # Publish the events in the default EventBridge bus
    logger.debug(f"Publishing to EventBridge: {events}")
    if events:
        eventbridge.put_events(
            Entries=events
        )

    return

def decode_kinesis_record(record):
    """
    Decodes a CDC record and returns a python object
    """
    payload = base64.b64decode(record["kinesis"]["data"])
    data = json.loads(payload)
    return data

def get_stock(product_ids, attempt=1):
    """
    Reads the aggregated stock in inventory of the products provided in the input parameter.
    Returns a dictionary where each key is the product id and the value is the quantity in stock.
    If the connection to the database is not successful, it will retry up to MAX_DB_READ_ATTEMPTS times.
    """
    global db_connection
    stock = {}

    if db_connection and attempt < MAX_DB_READ_ATTEMPTS:
        cursor = db_connection.cursor()
        ids = ",".join(map(str, product_ids))

        query = f"SELECT ProductId, SUM([Quantity])\
                    FROM [Production].[ProductInventory]\
                    WHERE [ProductId] IN ({ids})\
                    GROUP BY [ProductId]"
        logger.debug(f"Executing query: {query}")
        cursor.execute(query)
        row = cursor.fetchone()
        while row:
            stock[str(row[0])] = row[1]
            logger.debug(f"ProductId = {row[0]}, Inventory Quantity = {row[1]}")
            row = cursor.fetchone()
    elif attempt < MAX_DB_READ_ATTEMPTS:
        logger.debug("Attempting to reconnect to the database...")
        db_connection = connect_to_database()
        get_stock(product_ids=product_ids, attempt=attempt+1)

    return stock

def create_event(record, source):
    """
    This function returns an EventBridge event based on the information within a record
    """
    # The detail of the event is a dictionary with any arbitrary structure.
    # In this case it contains all the information from the data and metadata items
    # provided by DMS, as it may be relevant for applications downstream.
    # Depending on the use case, removing certain fields might be needed for regulatory purposes.
    event = {
        "Time": parse(record["metadata"]["timestamp"]),
        "Source": source,
        "DetailType": **ProductStockOverbooked**,
        "Detail": json.dumps({
            "data": record["data"],
            "metadata": record["metadata"]
        })
    }

    return event

This piece of code also uses external packages, so similarly, edit the requirements.txt file under the transactions directory and paste the following text. Note that pymssql uses native extensions, so the SAM CLI feature to use a Lambda-compatible container will come handy.
```
pymssql>=2.2.3
python-dateutil
```

At this step, fill in the template.yaml file with the following SAM template. I'll highlight a couple of points I think of interest:

Each Lambda function uses a different architecture: arm64, which runs on AWS Graviton processors and is very cost-performance effective, and x86_64. The reason for the latter is that, at the time of writing, there are no ARM compatible versions for the pymssql package.
Note that there are no explicit IAM roles defined in the template. SAM will create a role with basic permissions for Lambda functions if you don't specify an explicit one. The nice thing is that you can also add other policies to that implicit role. You can use SAM policy templates, AWS managed policies, customer managed policies, or you can define your own inline. In this case, I opted for two SAM policy templates EventBridgePutEventsPolicy and AWSSecretsManagerGetSecretValuePolicy, which make it straightforward to grant those self-described permissions to the functions.

AWSTemplateFormatVersion: 2010-09-09
Transform: AWS::Serverless-2016-10-31
Description: SAM Template for resources in the III unlocking innovation blog post

# Sets the timeout for all the functions in the application
Globals:
Function:
    Timeout: 10

Parameters:
SourceStreamName:
    Type: String
    Default: mssql-cdc
    Description: ARN of the Kinesis data stream to use as source
VpcId:
    Type: AWS::EC2::VPC::Id
    Description: Id of the VPC that has connectivity to the source database. A VPC Endpoint for EventBridge will be deployed there.
SubnetIds:
    Type: List<AWS::EC2::Subnet::Id>
    Description: List of SubnetIds that have connectivity to the source database. A VPC Endpoint for EventBridge will deploy an ENI on each subnet.
VpcEndpointSecurityGroupIds:
    Type: List<AWS::EC2::SecurityGroup::Id>
    Description: List of SecurityGroupIds for the VPC Endpoint for EventBridge.
SecurityGroupIds:
    Type: List<AWS::EC2::SecurityGroup::Id>
    Description: List of SecurityGroupIds that allow connecting to the source database
DatabaseConnectionSecretArn:
    Type: String
    Description: ARN of the AWS SecretsManager secret containing connection details for the source database
DatabaseName:
    Type: String
    Default: AdventureWorks2019
    Description: Name of the source database from where to read the current inventory

Resources:
# This VPC endpoint is needed because the TransactionHistoryProcessingFunction will have ENIs deployed in private subnets
EventBridgeVPCEndpoint:
    Type: AWS::EC2::VPCEndpoint
    Properties:
    PrivateDnsEnabled: true
    SecurityGroupIds: !Ref VpcEndpointSecurityGroupIds
    ServiceName: !Sub com.amazonaws.${AWS::Region}.events
    SubnetIds: !Ref SubnetIds
    VpcEndpointType: Interface
    VpcId: !Ref VpcId

ProductReviewProcessingFunction:
    Type: AWS::Serverless::Function
    Properties:
    Description: This function processes records from the CDC-populated Kinesis data stream related to the ProductReview table that match certain conditions
    CodeUri: reviews/
    Handler: app.lambda_handler
    Runtime: python3.9
    Architectures:
        - arm64
    Events:
        KinesisSourceStream:
        Type: Kinesis
        Properties:
            Stream: !Sub
                - "arn:${AWS::Partition}:kinesis:${AWS::Region}:${AWS::AccountId}:stream/${streamName}"
                - streamName: !Ref SourceStreamName
            StartingPosition: TRIM_HORIZON
            FilterCriteria:
            Filters:
                - Pattern: '{
                    "data": {
                        "metadata": { 
                        "operation": ["update", "insert"],
                        "schema-name": ["Production"],
                        "table-name": ["ProductReview"],
                        "record-type": ["data"]
                        },
                        "data": {
                        "Rating": [ {"numeric": ["<", 4]}]
                        }
                    }
                    }'
    Policies:
        - EventBridgePutEventsPolicy:
            EventBusName: default

TransactionHistoryProcessingFunction:
    Type: AWS::Serverless::Function
    Properties:
    Description: This function processes records from the CDC-populated Kinesis data stream related to the TransactionHistory table that match certain conditions
    CodeUri: transactions/
    Handler: app.lambda_handler
    Runtime: python3.9
    Architectures:
        - x86_64
    VpcConfig:
        SecurityGroupIds: !Ref SecurityGroupIds
        SubnetIds: !Ref SubnetIds
    Environment:
        Variables:
        DB_SECRET_ARN: !Ref DatabaseConnectionSecretArn
        DB_NAME: !Ref DatabaseName
    Events:
        KinesisSourceStream:
        Type: Kinesis
        Properties:
            Stream: !Sub
                - "arn:${AWS::Partition}:kinesis:${AWS::Region}:${AWS::AccountId}:stream/${streamName}"
                - streamName: !Ref SourceStreamName
            StartingPosition: TRIM_HORIZON
            FilterCriteria:
            Filters:
                - Pattern: '{
                    "data": {
                        "metadata": { 
                        "operation": ["update", "insert"],
                        "schema-name": ["Production"],
                        "table-name": ["TransactionHistory"],
                        "record-type": ["data"]
                        },
                        "data": {
                        "TransactionType": ["S"]
                        }
                    }
                    }'
    Policies:
        - EventBridgePutEventsPolicy:
            EventBusName: default
        - AWSSecretsManagerGetSecretValuePolicy:
            SecretArn: !Ref DatabaseConnectionSecretArn

Now it's time to build the template and deploy the solution. Make sure the AWS CLI is configured with an AWS principal with the proper permissions and you're using a profile attached to the AWS account you want to use. Note that SAM does some bootstrapping upon the first use (e.g.: creating an S3 bucket for uploading the processed templates). For the deployment step, you'll need to introduce some parameters such as the name of the application and some CloudFormation behaviors, but the SAM CLI has the --guidedoption that helps you introduce those parameters and store them locally for further reuse. Run the following commands on a terminal
```
sam build --use-container
sam deploy --guided
```

In subsequent calls after SAM CLI has installed the dependencies locally and you have set the parameters, you can simply run sam build && sam deploy (unless you want to change any of the two).

Conclusion

Following this post, you now have a steady stream of business events that you can use to build event-driven applications that react to them. In the next blog post, I'll explore how to use the DMS and the existing CDC stream to build an advanced use case.

Clean up

You can delete the resources you have deployed using SAM CLI with the following command:

sam delete

Unlock innovation using data from your on-premises databases (II) - Streaming state changes

Tasio Guevara — Mon, 14 Feb 2022 18:05:19 +0000

In my last blog post, I explained the motivation to continuously extract data from relational databases on-premises and introduced Change Data Capture (CDC) and AWS Database Migration Service (DMS) as tools to do this while minimizing the impact on the database.

In this blog post, I show how to implement a solution to get data changes into a data stream. Data streams are a powerful tool to build near real-time analytics and other use cases, such as building applications that react to state changes. For example, you may want to trigger a process when a customer updates their consent to receive marketing information or updates their email address.

In this solution, I will use AWS DMS to continuously replicate data from a SQL Server database into an Amazon Kinesis data stream. In this data stream, each record will contain the row that was changed (updated, inserted, or deleted) in the source database with all the current and previous values for each field.

Below, you can find a diagram that represents the architecture of the solution for a database on-premises and the architecture I used on my lab environment to prepare this blog post.

Because you need network connectivity between AWS DMS and your source database, I have, simply for illustrative purposes, placed a site-to-site VPN connection between the on-premises local network and a VPC in AWS. Note that there are several alternatives to establish connectivity between on-premises networks and AWS. In my lab setup, I am running the source database in a VPC, so I just need to run my AWS DMS replication instance in the same VPC or, better yet, in another VPC with connectivity to the former one (using, for example, VPC peering).

Figure 1 - Architecture for the solution with a database on-premises

Figure 2 - Architecture for the lab environment

Sample database

For this exercise, as a sample database, I will use a usual suspect in the SQL Server world, the AdventureWorks sample database. More precisely, I’ll use the 2019 OLTP (Online Transaction Processing) version, as I want to simulate an OLTP workload and I want will be using SQL Server 2019 running on Amazon RDS.

I will not dive into the details about how to deploy such an environment, but if you want to replicate it, take into account that in order to restore a database from a native SQL Server backup (like the one you'd download from the Microsoft AdventureWorks web site), you need to perform the steps described in the How do I perform native backups of an Amazon RDS DB instance that's running SQL Server? article.

Preparing the database

The sample database contains five schemas and a total of 68 tables (not counting administrative tables and the ‘dbo’ schema). For this exercise, I’m going to use three tables in the ‘Production’ schema: ‘ProductInventory’, ‘ProductReview’, and ‘TransactionHistory’.

To enable continuous replication, you need to configure the database. You can find prerequisites and detailed instructions in Using a Microsoft SQL Server database as a source for AWS DMS. In my case, because I’m using AWS RDS, so I do the following:

Execute this query in SQL Server to enable CDC on the database:
```
exec msdb.dbo.rds_cdc_enable_db 'AdventureWorks2019'
```

Execute these queries in SQL Server to enable CDC on each table:

USE [AdventureWorks2019]
exec sys.sp_cdc_enable_table
@source_schema = N'Production',
@source_name = N'ProductInventory',
@role_name = NULL,
@supports_net_changes = 1
GO

USE AdventureWorks2019
exec sys.sp_cdc_enable_table
@source_schema = N'Production',
@source_name = N'ProductReview',
@role_name = NULL,
@supports_net_changes = 1
GO

USE AdventureWorks2019
exec sys.sp_cdc_enable_table
@source_schema = N'Production',
@source_name = N'TransactionHistory',
@role_name = NULL,
@supports_net_changes = 1
GO

Run these queries to configure and enable the retention time (in seconds) for the changes in the transaction log. The documentation recommends 86399 seconds (one day), but you need to consider your need and the implications of having, potentially, to store that data in your local disks.
```
USE AdventureWorks2019
EXEC sys.sp_cdc_change_job @job_type = 'capture', @pollinginterval = 86399
EXEC sp_cdc_stop_job 'capture'
EXEC sp_cdc_start_job 'capture'
```

If you are curious about the effect of these commands, you can browse the SQL Server system tables and check that now there are a few tables under a schema named ‘cdc’.

Prerequisites for AWS DMS

There are a few things that you need in your environment to successfully run a migration task in AWS DMS.

You can find more detailed instructions about how to set up some resources in the Prerequisites section of the AWS DMS Documentation.

An IAM principal that has permissions to use AWS DMS, create the IAM roles in the next point, and full access to Kinesis data streams.
Two IAM roles to use DMS from the AWS CLI. You can create them following https://docs.aws.amazon.com/dms/latest/userguide/CHAP_Security.html#CHAP_Security.APIRole
A VPC with:
- At least two subnets that have a route (and allow traffic) to your database server.
- If your subnet is private, you will need either a NAT Gateway or VPC endpoints for Amazon Kinesis and AWS Secrets Manager, so the migration task can reach their corresponding endpoints.
- A Security Group that allows outbound traffic to your database server on the right port (1433 by default for SQL Server).

Setting up AWS DMS

AWS DMS requires the following components to run a replication task:

Source and target endpoints, each with associated IAM roles to access the necessary AWS resources (if any).
A replication instance running in a VPC that can connect to those endpoints.

Configuring the source endpoint

Storing connection details in AWS Secrets Manager

You need to configure the endpoint with the details needed to connect to your source database. You will typically need some user name and password for that, which is sensitive information. AWS DMS supports storing that sensitive data in AWS Secrets Manager. Since this is a best practice, first create a secret in AWS Secrets Manager containing the following key/value pairs:

{
  "host": "<ip-address-or-url-of-your-database-server>",
  "port": 1433,
  "username": "<your-username>",
  "password": "<your-password>"
}

Save the previous JSON object replacing the values between angle-brackets (<>) with your own in a file named sourcedb-creds.json.

Run the following command using the AWS CLI and take note of the ARN of the secret you just created.

aws secretsmanager create-secret \
    --name sourcedb-credentials \
    --description "Credentials for the source SQL Server database" \
    --secret-string file://sourcedb-creds.json

Delete the previous file to avoid having your password stored in plain text around.

Note: if you don’t provide an explicit AWS Key Management Service (KMS) key, AWS Secrets Manager will use the default KMS Key to encrypt this secret. This key’s alias is “aws/secretsmanager”, which is what you will later use to grant AWS DMS access to it .

Creating a role for AWS DMS to access the secret

To access that secret, AWS DMS needs an IAM role that grants access to it and to the KMS key that AWS Secrets Manager used to encrypt it. In this section you are going to create an IAM policy that grants minimal permissions to those resources and an IAM role that uses that policy.

Using your favorite editor, create a text file and name it read-sourcedb-secret-policy.json; paste the following text substituting the values between angle brackets (<>) for your your own and save it.

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "AllowGetSecretValue",
            "Effect": "Allow",
            "Action": "secretsmanager:GetSecretValue",
            "Resource": "<your-secret-arn>"
        },
        {
            "Effect": "Allow",
            "Action": [
                    "kms:Decrypt",
                    "kms:DescribeKey"
                    ],
            "Resource": "arn:aws:kms:<your-region>:<your-account-id>:alias/aws/secretsmanager"
    }
    ]
}

Create the IAM policy running this AWS CLI command.

aws iam create-policy \
    --policy-name read-sourcedb-secret \
    --policy-document file://read-sourcedb-secret-policy.json \
    --description 'Policy that allows reading the sourcedb secret in AWS Secrets Manager'

Now you can create the IAM role. Every IAM role requires a trust policy that determines which principal(s) will be able to assume it. In this case, that principal will be the AWS DMS service. Create a new text file, paste the following content and save it as dms-trust-policy.json.

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Principal": {
                "Service": "dms.<your-region>.amazonaws.com"
            },
            "Action": "sts:AssumeRole"
        }
    ]
}

Creating the endpoint

Now you can create the endpoint.

Create a text file, paste the following content replacing the values between angle brackets (<>) with your own, and save it as sql-server-settings.json

{
    "DatabaseName": "AdventureWorks2019",
    "SecretsManagerAccessRoleArn": "arn:aws:iam::<your-account-id>:role/dms-secretsmanager-role",
    "SecretsManagerSecretId": "arn:aws:secretsmanager:<your-region>:<your-account-id>:secret:<your-secret-id>"
}

Run the following AWS CLI command:

Note: if you are using a VPC endpoint to access AWS Secrets Manager and don’t enable private DNS hostnames in it, you will need to add the parameter --extra-connection-attributes secretsManagerEndpointOverride=[vpce-09d9c3f89fxxxxxxx-abcd0ef1.secretsmanager.<your-region>.vpce.amazonaws.com](http://vpce-09d9c3f89fxxxxxxx-abcd0ef1.secretsmanager.<your-region>.vpce.amazonaws.com/)to the command. Replace the value with the endpoint DNS of your VPC endpoint, which you can find under the endpoint details in the VPC console.
```
aws dms create-endpoint \
    --endpoint-identifier source-sqlserver \
    --endpoint-type source \
    --engine-name sqlserver \
    --microsoft-sql-server-settings file://sql-server-settings.json
```

Take note of the ARN of this endpoint, you will need it when creating the replication task.

Creating the AWS DMS replication instance

To connect to the source database, the replication instance needs a subnet group. This group is a configuration that tells AWS DMS where to deploy Elastic Network Interfaces (ENI) so it can access resources in a VPC (e.g.: a VPC endpoint, an RDS instance, a VPN gateway, etc.). AWS DMS requires at least two subnets for a subnet group.

Get the identifiers of two subnets within your VPC, you can find them using the AWS Management Console or using the AWS CLI with this command:
```
aws ec2 describe-subnets --query "Subnets[?VpcId=='<your-vpc-id>'].SubnetId"
```

Create the subnet group using the following AWS CLI command:

aws dms create-replication-subnet-group \
--replication-subnet-group-identifier mssql-replication-subnet-group \
--replication-subnet-group-description "Subnet group for the SQL Server replication instance" \
--subnet-ids "<subnet-id-1>" "<subnet-id-2>"

Create the replication instance using the following AWS CLI command and take note of the instance ARN.

aws dms create-replication-instance \
    --replication-instance-identifier mssql-replication-instance \
    --allocated-storage 50 \
    --replication-instance-class dms.t3.small \
    --replication-subnet-group-identifier mssql-replication-subnet-group \
    --no-multi-az \
    --no-publicly-accessible

Note that I have set the storage to 50 GB, used a dms.t3.small instance, and used the -—no-multi-az flag. Depending on your needs, you may want to change these settings or additional ones (see the CreateReplicationInstance API documentation). You can find more guidance in Choosing the right AWS DMS replication instance for your migration.

Finally, I haven’t specified any security group, so the instance will use the default security group in the VPC. If you haven’t changed that security group, it will contain an inbound rule that allows all traffic coming from the same security group, and all outbound traffic anywhere.

Now, wait until the instance is status Available (you can check this in the AWS DMS console) and test the connection.

Figure 3: Testing the endpoint

Configuring the target endpoint

The target endpoint requires a Kinesis data stream and a role that allows DMS to access that Kinesis data stream.

Creating the Amazon Kinesis data stream

In this post, I am using a Kinesis data in provisioned mode with one shard as I am not expecting any spiky load. Depending on your own scenario, you might want to use on-demand mode or more shards in provisioned mode.

To create the data stream, run the following AWS CLI command.

aws kinesis create-stream \
    --stream-name mssql-cdc \
    --shard-count 1 \
    --stream-mode-details StreamMode=PROVISIONED

Granting access from AWS DMS to the Kinesis data stream

AWS DMS needs to be explicitly given access to the data stream you just created. For this, you need to create an IAM policy and attach it to a new role that can be assumed by the AWS DMS service.

Create a new text file named write-to-stream-policy.json, paste the following content, and save it.

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "AllowWritingToStream",
            "Effect": "Allow",
            "Action": [
                "kinesis:DescribeStream",
                "kinesis:PutRecord",
                "kinesis:PutRecords"
            ],
            "Resource": "arn:aws:kinesis:<your-region>:<your-account-id>:stream/mssql-cdc"
        }
    ]
}

Create the IAM policy using this AWS CLI command.

aws iam create-policy \
    --policy-name write-to-cdc-stream \
    --policy-document file://write-to-stream-policy.json \
    --description 'Policy that allows writing to the mssql-cdc Kinesis data stream'

Create the IAM role that AWS DMS will assume running the following AWS CLI command. Note that this role uses the same trust policy as the IAM role used for the source endpoint.

aws iam create-role \
    --path / \
    --role-name dms-kinesis-role \
    --assume-role-policy-document file://dms-trust-policy.json \
    --description 'Role for DMS to access Amazon Kinesis'

Attach the policy to the newly created role by running this command.

aws iam attach-role-policy \
    --role-name dms-kinesis-role \
    --policy-arn arn:aws:iam::<your-account-id>:policy/write-to-cdc-stream

Creating the target endpoint

Now, you have everything set up to create the target endpoint. For that, run this command.

aws dms create-endpoint \
    --endpoint-identifier target-kinesis \
    --endpoint-type target \
    --engine-name kinesis \
    --kinesis-settings \
StreamArn=arn:aws:kinesis:<your-region>:<your-account-id>:stream/mssql-cdc,\
ServiceAccessRoleArn=arn:aws:iam::<your-account-id>:role/dms-kinesis-role,\
MessageFormat=JSON

As with the source endpoint, test the connectivity to make sure there are no issues. If there are issues, make sure that the subnets associated with the replication instance have access to Amazon Kinesis. Some hints if you are using private subnets and find problems

Check there is a VPC endpoint for Kinesis with ENIs deployed on those subnets.
Check that the VPC endpoint has enabled private DNS hostnames in it. If not, you will need to add the parameter --extra-connection-attributes with the private DNS hostname of the VPC endpoint to the command to create the endpoint.
Alternatively, you can use a NAT gateway or another type of proxy to access the Internet.

Setting up the replication task

At this point, you have everything ready to create and launch a replication task.

A replication task needs to have a table mapping, so it knows which tables and columns from the data source to read and how to write them to the target. You can also define a series of (optional) task settings you can use to customize its behavior. With those settings you can configure multiple aspects, such as logging, data validation, time travel, or data filtering. You can find a list of the available settings in the Specifying task settings for AWS Database Migration Service tasks documentation.

In this example, you are going to configure a task that does continuous replication of two source tables in the SQL Server database into the Kinesis data stream. In the output, each record will include the updated and previous values for each row affected (when relevant, as for deletes and inserts this does not apply).

Depending on the database operation Each record in the stream will represent and updated row from the database with a JSON dictionary that includes three items: “data”, “before”, and “metadata”.

The “data” and “before” items will contain a key-value pair for column and value key for each column in the table, where “data” has the values in the updated row, and “before” has the values in the updated row prior to the database operation.

The “metadata” item will contain information about the operation, such as its timestamp, the operation type (update, delete, or insert), or the table and schema the operation was performed on. Note that when inserting or deleting a row, the “before” element will not be included in the dictionary.

Here’s a sample record for an update operation:

{
    "data": {
        "ProductReviewID": 3,
        "ProductID": 937,
        "ReviewerName": "Alice",
        "ReviewDate": "2013-11-15T00:00:00Z",
        "EmailAddress": "alice@example.com",
        "Rating": 3,
        "Comments": "Maybe it's just because I'm new to mountain biking, but I had a terrible time getting use\r\nto these pedals. In my first outing, I wiped out trying to release my foot. Any suggestions on\r\nways I can adjust the pedals, or is it just a learning curve thing?",
        "ModifiedDate": "2013-11-15T00:00:00Z"
    },
    "before": {
        "ProductReviewID": "3",
        "ProductID": "937",
        "ReviewerName": "Alice",
        "ReviewDate": "2013-11-15T00:00:00Z",
        "EmailAddress": "alice@example.com",
        "Rating": "2",
        "Comments": "Maybe it's just because I'm new to mountain biking, but I had a terrible time getting use\r\nto these pedals. In my first outing, I wiped out trying to release my foot. Any suggestions on\r\nways I can adjust the pedals, or is it just a learning curve thing?",
        "ModifiedDate": "2013-11-15T00:00:00Z"
    },
    "metadata": {
        "timestamp": "2022-02-02T11:52:59.913277Z",
        "record-type": "data",
        "operation": "update",
        "partition-key-type": "schema-table",
        "schema-name": "Production",
        "table-name": "ProductReview",
        "transaction-id": 128599
    }
}

Table Mapping

Diving a bit deeper into table mapping, you need to select which tables from which schemas are going to be read from the data source. You have the option of using the ‘%’ as a wildcard for the name, or part of it, of the tables and schemas. You can also and apply some transformations prior to data being written at the target. You can do things like filtering values, adding/removing columns, renaming, adding suffixes, or changing data types (you can find a complete list of actions and capabilities in the Transformation rules and actions documentation). In this example, I’m not applying any transformation, only reading two tables from two different schemas.

You can define those mappings either in the AWS DMS console or by using a JSON file. In this guide, I’m using the latter.

Create a text file and paste the following JSON object. Save the file as table-mapping-rules.json.

{
    "rules": [
        {
            "rule-type": "selection",
            "rule-id": "1",
            "rule-name": "products",
            "object-locator": {
                "schema-name": "Production",
                "table-name": "Product"
            },
            "rule-action": "include",
            "filters": []
        },
        {
            "rule-type": "selection",
            "rule-id": "2",
            "rule-name": "transaction-history",
            "object-locator": {
                "schema-name": "Production",
                "table-name": "TransactionHistory"
            },
            "rule-action": "include",
            "filters": []
        },
        {
            "rule-type": "selection",
            "rule-id": "3",
            "rule-name": "product-reviews",
            "object-locator": {
                "schema-name": "Production",
                "table-name": "ProductReview"
            },
            "rule-action": "include",
            "filters": []
        }
    ]
}

Task settings

AWS DMS allows defining the behavior of a task very granularly, although in this case I’m only going to define rules for logging into Amazon CloudWatch Logs and a behavior to include the “before” values of a row that has been updated.

Create a text file, paste the following JSON object, and save it as task-settings.json.

{
    "Logging": {
        "EnableLogging": true,
        "LogComponents": [
            {
                "Id": "TRANSFORMATION",
                "Severity": "LOGGER_SEVERITY_DEFAULT"
            },
            {
                "Id": "SOURCE_UNLOAD",
                "Severity": "LOGGER_SEVERITY_DEFAULT"
            },
            {
                "Id": "IO",
                "Severity": "LOGGER_SEVERITY_DEFAULT"
            },
            {
                "Id": "TARGET_LOAD",
                "Severity": "LOGGER_SEVERITY_DEFAULT"
            },
            {
                "Id": "PERFORMANCE",
                "Severity": "LOGGER_SEVERITY_DEFAULT"
            },
            {
                "Id": "SOURCE_CAPTURE",
                "Severity": "LOGGER_SEVERITY_DEFAULT"
            },
            {
                "Id": "SORTER",
                "Severity": "LOGGER_SEVERITY_DEFAULT"
            },
            {
                "Id": "REST_SERVER",
                "Severity": "LOGGER_SEVERITY_DEFAULT"
            },
            {
                "Id": "VALIDATOR_EXT",
                "Severity": "LOGGER_SEVERITY_DEFAULT"
            },
            {
                "Id": "TARGET_APPLY",
                "Severity": "LOGGER_SEVERITY_DEFAULT"
            },
            {
                "Id": "TASK_MANAGER",
                "Severity": "LOGGER_SEVERITY_DEFAULT"
            },
            {
                "Id": "TABLES_MANAGER",
                "Severity": "LOGGER_SEVERITY_DEFAULT"
            },
            {
                "Id": "METADATA_MANAGER",
                "Severity": "LOGGER_SEVERITY_DEFAULT"
            },
            {
                "Id": "FILE_FACTORY",
                "Severity": "LOGGER_SEVERITY_DEFAULT"
            },
            {
                "Id": "COMMON",
                "Severity": "LOGGER_SEVERITY_DEFAULT"
            },
            {
                "Id": "ADDONS",
                "Severity": "LOGGER_SEVERITY_DEFAULT"
            },
            {
                "Id": "DATA_STRUCTURE",
                "Severity": "LOGGER_SEVERITY_DEFAULT"
            },
            {
                "Id": "COMMUNICATION",
                "Severity": "LOGGER_SEVERITY_DEFAULT"
            },
            {
                "Id": "FILE_TRANSFER",
                "Severity": "LOGGER_SEVERITY_DEFAULT"
            }
        ]
    },
    "BeforeImageSettings": {
        "EnableBeforeImage": true,
        "FieldName": "before",
        "ColumnFilter": "all"
    }
}

Creating the replication task

Run the following AWS CLI commands to create and start the task respectively.

aws dms create-replication-task \
    --replication-task-identifier mssql-to-kinesis-cdc
    --source-endpoint-arn arn:aws:dms:<your-region>:<your-account-id>:endpoint:<your-source-endpoint-id> \
    --target-endpoint-arn arn:aws:dms:<your-region>:<your-account-id>:endpoint:<your-target-endpoint-id> \
    --replication-instance-arn arn:aws:dms:<your-region>:<your-account-id>:rep:<your-replication-instance-id> \
    --migration-type cdc \
    --table-mappings file://table-mapping-rules.json \
    --replication-task-settings file://task-settings.json \
    --resource-identifier mssql-to-kinesis-cdc

aws dms start-replication-task \
    --replication-task-arn arn:aws:dms:<your-region>:<your-account-id>:task:<your-task-id> \
    --start-replication-task-type start-replication

Note that, because this is a continuous replication task, you can decide at what point in time in the database you want to start the replication from and when to stop. You can also stop and resume the task.

After a few minutes, your task should be in the state “Replication ongoing“.

Figure 4: Task state

Testing the solution

To test the solution, perform some update, insert or delete in one of the two source tables, for example, run the following SQL statement against your source database:

UPDATE AdventureWorks2019.Production.ProductReview
SET Rating = 5 WHERE ProductReviewID = 2

Wait for about a minute until the replication task has written into your Kinesis data stream and then read from the stream.

If you are running this tutorial from a Unix-type command processor such as bash, you can automate the acquisition of the shard iterator using a nested command, like this (scroll horizontally to see the entire command):

SHARD_ITERATOR=$(aws kinesis get-shard-iterator --shard-id shardId-000000000000 --shard-iterator-type TRIM_HORIZON --stream-name <your-stream-name> --query 'ShardIterator')
aws kinesis get-records --shard-iterator $SHARD_ITERATOR --query "Records[-1].Data" | sed 's/"//' | base64 --decode

If you are running this tutorial from a system that supports PowerShell, you can automate acquisition of the shard iterator using a command such as this (scroll horizontally to see the entire command):

aws kinesis get-records --shard-iterator ((aws kinesis get-shard-iterator --shard-id shardId-000000000000 --shard-iterator-type TRIM_HORIZON --stream-name Foo).split('"')[4])`

Conclusion

In this post, I have shown how to set up a replication task to stream changes in a relational databases into a data stream. In the next post, I will showcase how to process data from the stream in real time.

Photo by Michael Dziedzic on Unsplash

Unlock innovation using data from your on-premises databases (I)

Tasio Guevara — Thu, 28 Oct 2021 06:47:30 +0000

Data has arguably become one of the most important assets for any business. Data captures the past and present state of an organization and can help shaping and forecasting its future.

Thanks to the advancements on analytics and artificial intelligence, and the doors that these technologies open for innovation, data has taken the center stage for many organizations’ strategy.

In this series of blog posts I am going to describe a few use cases that you can implement with minimal impact to your existing databases and with no need to migrate them to the cloud.

This first post explains the main cause that limits data-driven innovation and explains an approach to solve it. The posts to follow will dive deeper into the technical implementation of different solutions for different use cases.

Legacy database architectures

Even today, many businesses still struggle to use their existing data for new use cases other than those for which those databases were designed. There are several reasons for this, but the main one is due to the legacy database architectures that were used when setting those systems up.

A typical architecture consists on a transactional database, also known as OLTP (On-Line Transactional Processing)
database, and an analytical one or OLAP (On-Line Analytics Processing). Data is copied in batches from the OLTP to the OLAP database using some Extract-Transform-Load (ETL) processes. These processes are executed on a schedule (e.g.: once a day) and the tools are often provided by the same vendors of the database engines. In order to make this work, these databases are fine-tuned to achieve the required needs in terms of performance, availability, etc. This entails the work of developers, database administrators (DBAs), and business analysts to, among other tasks, design the right tables, indexes, and partitions; optimize transactional queries; and schedule maintenance and ETL jobs to minimize the impact on business operations.

To those challenges, you also need to add the inherent ones of scaling relational databases (TL;DR: it’s hard) and the constraints of using on-premises infrastructure, where you can’t just simply add more storage or spin up a more powerful server in a matter of seconds. The “you’re gonna need a bigger boat” scenario when operating on-premises usually means waiting for months until your database is running on an appropriately-sized server.

As you can rightfully guess, accommodating new uses cases in that ecosystem requires plenty of planning. Furthermore, if those use cases go into the direction of near real-time analytics or machine learning, you may find that your legacy databases are simply not fitting the bill (and rightfully so, as those databases were not designed for it).

Migrate to the cloud you say?

If you have kept with me so far, you may be thinking: “right, so I need to migrate those databases to the cloud, but...”. Well, that is an option with its own benefits, but I am aware that not everyone is ready or willing to migrate their databases to AWS.

As I explained in the introduction, the goal of this blog series is to show you how you can start leveraging the flexibility and scale of AWS to build your innovative use cases while keeping your databases where they are and minimizing the operational impact on them. One of the solutions for doing this is called Change Data Capture or CDC.

Change Data Capture

To define Change Data Capture (CDC), I’ll simply quote its article in Wikipedia:

In databases, change data capture is a set of software design patterns used to determine and track the data that has changed so that action can be taken using the changed data.

In a nutshell, it allows you to know that "something" has happened shortly after it has happened.

Although it depends on the source database engine, CDC typically uses the database transaction logs, so there is no need to query the database to extract the data. This has the benefit of having much lower impact on the database server than continuously reading from the tables you are interested in.

Implementing such design patterns is not an easy task, but don't fret, as there are offerings that provide out-of-the-box support for CDC.

There are many options, including AWS partners that have solutions for CDC in the AWS Partner Network and AWS' own AWS Database Migration Service (DMS).

AWS Database Migration Service

In this series, I will use AWS Database Migration Service (DMS).

AWS DMS is a cloud service that makes it easy to migrate relational databases, data warehouses, NoSQL databases, and other types of data stores. You can use AWS DMS to migrate your data into the AWS Cloud or between combinations of cloud and on-premises setups.

But more importantly (I promised you don't need to move your database from where it is), AWS DMS supports CDC and can run continuous replication jobs, so you can uninterruptedly copy data from a source database to different target databases and AWS services.

AWS DMS can use many of the most popular data engines as a source for data replication, so it's very likely that you can use it with your existing databases.

The option to use different and heterogeneous targets is the key for implementing different use cases flexibly. For example, you may want to use an Amazon Kinesis Data Stream for near real-time data processing, or Amazon S3 to hydrate a data lake that stores data for analytics or training machine learning models.

Conclusion

In this post I have outlined the issues with traditional database architectures when you need to implement new use cases and have described CDC as a way to overcome those limitations. Finally, I have provided a glimpse on some of the AWS DMS capabilities that are relevant for using CDC.

In the next blog post I will provide the technical implementation of a solution to enable CDC using AWS DMS and a Microsoft SQL Server source database.

Photo by Michael Dziedzic on Unsplash