DEV Community: omarrosadio

Streaming data from Amazon QLDB to OpenSearch

omarrosadio — Sun, 25 Sep 2022 06:09:08 +0000

The goal of this tutorial is to understand and implement a solution that streams data changes made on a ledger on Amazon QLDB to an OpenSearch Domain.
A common use case for this architecture would be a scenario where is required an event-driven architecture to process data and perform analytics in near real-time for data stored on a ledger on QLDB.

Architecture
Considerations
Deploying the infrastructure using CloudFormation
Configuring Access Control in OpenSearch
Inserting data in the Ledger
Querying the data in OpenSearch
Conclusion

Architecture

The way it works is as follows:
1) There is a Stream associated with the QLDB ledger. Every change made on the ledger is captured in the stream which sends that data to a Kinesis Stream (note that the QLDB stream is near-real time)
2) Once the data is in Kinesis Stream, consumers can process it. In this case to move the data to the destination (OpenSearch domain) a Kinesis Firehose Delivery Stream is configured. The delivery stream has associated a data transformation lambda function to filter only the relevant events we want to insert into OpenSearch.

Considerations

The configuration settings are in a basic mode and most of them has the default value. For a production workload these configurations should be changed to select an optimal value. For example:
The OpenSearch domain is created in a Public Networking mode but for a production scenario the VPC mode offers more control and security.
The Kinesis Firehose delivery stream buffers are configured at the lowest value possible to see the data being moved to its destination as soon as possible, but for a production scenario these values should be carefully chosen.
OpenSearch Domain Master user and password are configured as paramaters in the CFN template. However, there are more secure alternatives such as using Parameter Store parameters.

Deploying the infrastructure using CloudFormation

We will create all the components showed in the architecture diagram using a CloudFormation template. After that, some manual configurations will be required.
The template can be found in this GitHub repository with the name streaming-from-qldb-to-opensearch.yaml.

The parameter values you would need to change are:
- OpenSearchMasterUsername: The master username you will use to connect to the OpenSearch service
- OpenSearchMasterPassword: The master password you will use to connect to the OpenSearch service. Passwords must contain at least one uppercase letter, one lowercase letter, one number, and one special character.

Create the Stack and before continue we need all components created. This can take up to 30 minutes (most of the time is consumed in the OpenSearch domain creation).

Configuring Access Control in OpenSearch

The IAM Role created that will be used by Kinesis Firehose to move the data to the OpenSearch Domain has the required IAM Permissions to do that (actions that includes es). However, there is required to explicitly allow and assign the permissions from the OpenSearch Domain. To do this we need to configure these permissions entering to the dashboard.

The dashboard URL can be obtained from the OpenSearch console or from the CloudFormation Stack - Outputs section with the name "OpenSearchDashboard":

Use the username and password entered as input parameters during Stack creation:

Then go to the hamburger menu in the top left corner -> "Security" -> "Roles":

For this demo, select the "all_access" role and in "Mapped users" section click on "Manage mapping":

And add the Kinesis Firehose Delivery Stream Role ARN in the "Backend roles" section. This value can be found in CloudFormation Stack - Outputs section with the name "KinesisFirehoseRole".

And click on Map button.

Inserting data in the Ledger

Now that everything is configured, we need to insert some sample data to see the solution in action.

Go to "Amazon QLDB" -> "PartiQL editor" on the web console. Select the recently created ledger (qldb-stream) and run the following script to create a table, an index and insert some data into the table.

CREATE TABLE Person;
CREATE INDEX ON Person (GovId);
INSERT INTO Person VALUE {'GovId':'AAA-BB-0001', 'GovIdType': 'Passport', 'FirstName':'Raul', 'LastName':'Lewis', 'Country': 'USA', 'Age': 23}; 
INSERT INTO Person VALUE {'GovId':'CCC-DD-0002', 'GovIdType': 'Passport', 'FirstName':'Brent', 'LastName':'Logan', 'Country': 'USA', 'Age': 40};
INSERT INTO Person VALUE {'GovId':'EEE-FF-0003', 'GovIdType': 'Passport', 'FirstName':'Alexis', 'LastName':'Pena', 'Country': 'USA', 'Age': 29};
INSERT INTO Person VALUE {'GovId':'GGG-HH-0004', 'GovIdType': 'Passport', 'FirstName':'Melvin', 'LastName':'Parker', 'Country': 'Mexico', 'Age': 18};
INSERT INTO Person VALUE {'GovId':'III-JJ-0005', 'GovIdType': 'DNI', 'FirstName':'Salvatore', 'LastName':'Spencer', 'Country': 'Mexico', 'Age': 21};
INSERT INTO Person VALUE {'GovId':'LLL-MM-0006', 'GovIdType': 'DNI', 'FirstName':'Carlos', 'LastName':'Trump', 'Country': 'Peru', 'Age': 42};
INSERT INTO Person VALUE {'GovId':'NNN-OO-0007', 'GovIdType': 'DNI', 'FirstName':'John', 'LastName':'Connor', 'Country': 'Chile', 'Age': 36};
INSERT INTO Person VALUE {'GovId':'PPP-QQ-0008', 'GovIdType': 'Other', 'FirstName':'Diana', 'LastName':'Brown', 'Country': 'Brazil', 'Age': 30};
INSERT INTO Person VALUE {'GovId':'RRR-SS-0009', 'GovIdType': 'Other', 'FirstName':'Albert', 'LastName':'Johnson', 'Country': 'Peru', 'Age': 29};
INSERT INTO Person VALUE {'GovId':'TTT-UU-0010', 'GovIdType': 'Other', 'FirstName':'Freddy', 'LastName':'Rose', 'Country': 'Ecuador', 'Age': 33};

The statements should be executed successfully:

Querying the data in OpenSearch

After the data is inserted in the ledger, it can take up to 1 minute to be send from QLDB Stream to Kinesis Data Stream. And then up to 1 minute from Kinesis Data Firehose to OpenSearch. So after about 2 minutes from data insertion it should be visible on OpenSearch.

To validate the index was successfully created go to the hamburger menu in the top left corner -> "Query Workbench" and execute the query:

SHOW tables LIKE %;

And an index with the name we use as an input parameter in the CFN template should appear:

If the index is not listed, then validate step 4 and make sure the role ARN was granted with the corresponding permissions.

To see the data on OpenSearch go to the hamburger menu in the top left corner -> "Visualize" and click on "Create index pattern"

On "Index pattern name" field enter the index name configured during Stack creation (OpenSearchIndexName parameter) and click on "Next step":

And finally click on "Create index pattern":

The inserted data on the ledger is visible on OpenSearch and new changes will also be streamed to OpenSearch. Go to the hamburger menu -> "Discover" section to see it:

Conclusion

Having the ability to analyze data in real or near real-time is crucial for analytics workloads. QLDB's supports this using Streams which sends the data to a Kinesis Stream. Once the data is on Kinesis Stream it can be processed in multiple ways. In this example it is delivered to a OpenSearh destination.
I hope this example could help and could be used as a starting point for custom implementations. With the base infrastructure more components can be added: more delivery streams consuming from Kinesis Data Stream to delivery in parallel to another targets e.g. S3, Redshift, New Relic, etc.

Implementing Change Data Capute (CDC) with Aurora Serverless v2

omarrosadio — Fri, 22 Apr 2022 20:46:57 +0000

Aurora Serverless v2 is now available after a long wait and it claim to solve many issues and limitations from its predecessor.
One of the limitations of Aurora Serverless v1 is the impossibility of using CDC (Change Data Capture) with AWS DMS (Database Migration Service) but now it is possible using the v2.
In this guide we will review step by step the configuration and implementation of CDC with Aurora Serverless v2 and AWS DMS.

Actually, the configuration is pretty much the same as configuring CDC on a provisioned Aurora DB Cluster. Also consider that all the configurations are based on us-east-1 (N. Virginia) region.

Creating the Database
As first step, we need to create the Parameter Group, Subnet Group and the Database cluster. I am using almost all the default parameters, changing only the required ones to enable CDC (those parameters are in italic letters).

Parameter Group
Click on the 'Create Parameter Group' button:

Select:
Parameter Group family: aurora-mysql8.0
(Currently it is the only one compatible with Aurora Serverless v2)

Type:
DB Cluster Parameter Group

Group Name:
pg-servelessv2-cdc

Description:
Enable change data capture

Once the Parameter group is created, select it and click on 'Edit parameters':

We need to change these parameters:
binlog_format: ROW
binlog_checksum: NONE

And save changes:

Subnet group
Using subnet group we can specify in which subnets the database will be deployed. So in this case I will select public subnets to be able to connect to the DB easily.

Click on the 'Create DB subnet group' button:

Name:
sg-db-publicaccess
Description:
Using public subnets for demo only
VPC:
your_VPC_id
Availability Zones:
az_ids (is mandatory to select at least 2)

And now the subnet groups is created:

Database

Click on the 'Create Database' button:

Database creation method: Standard create
Engine type: Amazon Aurora
Edition: Amazon Aurora MySQL-Compatible Edition
Replication features: Single-master (default)
Engine version: Aurora MySQL 3.02.0 (compatible with MySQL 8.0.23)
Templates: Dev/Test
DB cluster identifier: db-test-cdc
Master username: admin
Master password: supersecretpassword
DB instance class: Serverless v2 - new
Capacity range - Minimum ACUs: 0.5
Capacity range - Maximum ACUs: 1
Multi-AZ deployment: Don't create an Aurora Replica
Virtual private cloud (VPC): your_VPC_id
Subnet group: sg-db-publicaccess (the subnet group previously created)
Public access: Yes
VPC security group: Create new
New VPC security group name: rds-publicaccess
Availability Zone: No preference
Database port: 3306
Database authentication options: Password authentication
Initial database name: sampledb
DB cluster parameter group: pg-servelessv2-cdc (the parameter group previously created)
DB parameter group: default.aurora-mysql8.0

Click on Create database (it takes approximately 15 minutes to create completely):

Also modifiy the created Security Group (rds-publicaccess) to allow inboud traffic on port 3306 from 0.0.0.0/0

Creating the S3 bucket
The next step is to create an S3 bucket which will be used as the target destination for the full load and change data capture task.
It does not required any special configuration, we can use a bucket with default creation parameters:
Name: test-dms-s3-target

Creating the DMS Replication Instance

We need to create the DMS replication subnet group. Similar to database subnet group, it helps to specify the subnets that will be used. In this case I am selecting private subnets but can be any with proper connectivity to the database instance and S3 bucket.
Name: sg-dms

Additionally, create the Security group that will be attached to the DMS replication instance. It requires allow outbound traffic.

And create the DMS Replication Instance:

Name: dms-instance
Instance class: dms.t3.small
Engine version: 3.4.6
Allocated storage: 20GB
VPC: The same VPC as the one RDs instance belongs to
Multi AZ: Dev or test worload
Publicly accessible: No
Replication subnet group: sg-dms (the previously created)
Availability zone: No preference
VPC security group(s): dms-replication-instance (the previously created)
KMS key: Default

Inserting sample data into the database
Before create the DMS task I am going to insert sample data into the database to validate the task works as expected. The database imported is from MySQL sample, specifically, the Sakila sample database. These are 2 scripts to execute so it is very quick to replicate.
Connect to the database using the user and password configured and setting the Database endpoint properly (in case of connection errors check the SG allow traffic on port 3306 for the IP range):

Then execute the downloaded scripts: sakila-schema.sql and sakila-data.sql

And now we have populated the database:

Creating the DMS Tasks
Create the Source Endpoint to connect to the database and the Target Endpoint to connect to the S3 bucket:

And test the connection:

Before create the Target Endpoint for the S3 bucket, we need to create an IAM role granting access to put/delete objects on the bucket:

Select DMS as the AWS service:

Don't select any policies for now:

Select an proper name and create the role:

And Add permissions -> Create inline policy

The policy is as following (change the bucket name for the corresponding value):

{
    "Version" : "2012-10-17",
    "Statement" : [ 
      {
        "Effect" : "Allow",
        "Action" : [
          "s3:PutObject",
          "s3:DeleteObject"
        ],
        "Resource" : "arn:aws:s3:::test-dms-s3-target/*"
      },
      {
        "Effect" : "Allow",
        "Action" : "s3:ListBucket",
        "Resource" : "arn:aws:s3:::test-dms-s3-target"
      } 
    ]
  }

Now, we can create the Target endpoint:

And test the connection:

To create the Database Migration Task we use the resources previously created:

Task identifier: full-load-and-cdc-auroraserverless

For Task setting in this demo we can leave the default parameters except for the Enable CloudWatch logs which will be helpful to validate everything is Ok:

For Table mappings we only need to specify the schema (sakila in this case) and leave the other parameters with default values:

And finally click on Create task:

Validating the CDC feature
Lets keep the task running for some minutes untill the state is "Load complete, replication ongoing":

Check from table statistics that the data was loaded successfully:

The relevant fields are Full load rows and Total rows:

If you navigate through the bucket objects, you will notice there is a folder for each table:

And for the first load is structured as follows:

To check the CDC in action, lets insert some rows:

INSERT INTO `sakila`.`customer` (`customer_id`, `store_id`, `first_name`, `last_name`, `email`, `address_id`, `active`, `create_date`, `last_update`) VALUES ('600', '2', 'JOHN', 'SMITH', 'JOHN.SMITH@sakilacustomer.org', '605', '1', '2006-02-14 22:04:37', '2006-02-15 04:57:20');
INSERT INTO `sakila`.`customer` (`customer_id`, `store_id`, `first_name`, `last_name`, `email`, `address_id`, `active`, `create_date`, `last_update`) VALUES ('601', '2', 'WILL', 'CARTER', 'WILL.CARTER@sakilacustomer.org', '605', '1', '2006-02-14 22:04:37', '2006-02-15 04:57:20');
INSERT INTO `sakila`.`customer` (`customer_id`, `store_id`, `first_name`, `last_name`, `email`, `address_id`, `active`, `create_date`, `last_update`) VALUES ('602', '2', 'DONALD', 'JACKSON', 'DONALD.JACKSON@sakilacustomer.org', '605', '1', '2006-02-14 22:04:37', '2006-02-15 04:57:20');

And also delete some rows from the customer table:

DELETE FROM `sakila`.`customer` WHERE (`customer_id` = '9');
DELETE FROM `sakila`.`customer` WHERE (`customer_id` = '15');
DELETE FROM `sakila`.`customer` WHERE (`customer_id` = '4');
DELETE FROM `sakila`.`customer` WHERE (`customer_id` = '433');
DELETE FROM `sakila`.`customer` WHERE (`customer_id` = '599');

After a couple of minutes (it could take 2 or 3 minutes), the changes should be reflected on both the task statistics and S3 bucket:

Conclusion
With the new version of Aurora serverless there is not limitation on setting up DMS migration tasks with Change Data Capture. It allows to take advantage of the serverless model and also the configuration is very similar to provisioned version.

Using AWS SDK V2 for Java to read DynamoDB cross-account

omarrosadio — Fri, 24 Dec 2021 03:03:15 +0000

For medium and large size companies, is usual to have multiple AWS accounts grouped by functionality, environment, business goals, etc.
Sometimes, when is required cross-account access for resources, could be a problem due to the number of steps involved to grant access. And in the case of DynamoDB, additionally steps appears (different from services like KMS where you configure cross-account access and it is enough, for services like DynamoDB it is needed first assume the role and then procede with API calls).

In this sample, I will describe step by step how to configure permissions to allow DynamoDB cross account access and test it using AWS SDK V2 for Java deployed in EC2 (however, the same concept could be applied to AWS Lambda).

Scenario
Access from an Java Application on Account "A" to and DynamoDB Table on Account "B".

Diagram of resources to be created

Considerations

Account owner of DynamoDB Tables will be referenced as "Account A".
Account in which Java Application will be deployed and need access to Tables from "Account A", will be reference as "Account B".
Complete Java Code can be found on this repository. It uses maven as dependency manager and AWS SDK V2 for Java.

Steps

1. Create Role on Account A (DynamoDB resources owner)

On account A, create a Role and choose "Another AWS Account" as type of trusted entity. Enter the Account ID from Account B:

For permissions policies, I am attaching "AmazonDynamoDBReadOnlyAccess" managed policy only for demo purposes, but on real usage it should follow the Principle of Least Privilege to limit access.

Create the role, choose a suitable role name:

Copy the Role ARN because it will be used in the next step:
arn:aws:iam::<ID_Account_B>:role/CrossAccountReadDynamo

2. Create policy on Account B with permissions to assume role created on step 1

On Account B, create a Policy to grant permissions to assume the role from Account A. For policy JSON use:
{ "Version": "2012-10-17", "Statement": { "Effect": "Allow", "Action": "sts:AssumeRole", "Resource": "arn:aws:iam::<ID_Account_A>:role/CrossAccountReadDynamo" } }

And choose a name for the policy:
assume_role_account_A

3. Create role on Account B and attach the policy created on step 2

On Account B, create a role and select EC2 as type of trusted entity:

Attach the policy recently created:

Name: EC2_AssumeRole_Account_A

4. Test cross account access through Java Application

The Java Application will be deployed on and EC2 Instance in Account B. This EC2 Instance needs to has attached the Role created previously.
The first step on code requires make an API Call to the Security Token Service (AWS STS) and in this way generate temporary credentials to be able to assume the role from Account A.
Modify the variables accordingly:

roleARN: substitute with Role ARN created on step 1
roleSessionName: an identifier to the temporary session
region: region where the DynamoDB resources are deployed

/*Change this variables according to the role created*/ String roleARN = "arn:aws:iam::<ACCOUNT_A>:role/<ROLE_NAME>"; String roleSessionName = "DynamoCrossAccount"; Region region = Region.US_EAST_1;

Once the temporary credentials are generated, is needed to set the AccessKey, SecretAccessKey and SessionToken:
AwsSessionCredentials awsCreds = AwsSessionCredentials.create(myCreds.accessKeyId(), myCreds.secretAccessKey(), myCreds.sessionToken());

With the new credentials configured, the subsequents API Calls will be used the role from Account A. So, now we can consume and do operations over DynamoDB Tables of Account A. To test it, I will list all the existing tables:



ListTablesResponse response = null;
                if (lastName == null) {
                    ListTablesRequest request = ListTablesRequest.builder().limit(10).build();

                    response = ddb.listTables(request);
                } else {
                    ListTablesRequest request = ListTablesRequest.builder().exclusiveStartTableName(lastName).build();
                    response = ddb.listTables(request);
                }

                List<String> tableNames = response.tableNames();

                if (tableNames.size() > 0) {
                    for (String curName : tableNames) {
                        System.out.format("* %s\n", curName);
                    }
                } else {
                    System.out.println("No tables found!");
                    System.exit(0);
                }

                lastName = response.lastEvaluatedTableName();
                if (lastName == null) {
                    moreTables = false;
                }

And the expected output should list resources from the other account. Achieving cross-account access:

Again, the complete code is on GitHub and it is based on Official AWS documentation, specifically: