In this video, you’ll see how to send change data capture (CDC) information from relational
databases to Amazon Kinesis Data Streams by using AWS Database Migration Service (AWS DMS).
With this capability, you can configure CDC replication and implement upstream
data changes to a data lake or data warehouse with near real-time capability.
This will allow you to get additional data insights by unlocking use cases
for analytics as well as artificial intelligence and machine learning.
In this demo we'll be using an Amazon Aurora MySQL-compatible database
as our upstream datastore.
We’ll use AWS DMS to send CDC information to Amazon Kinesis Data Streams.
Kinesis Data Streams is a massively scalable and durable data streaming
service that can collect and process large streams of data records in near real-time.
This allows upstream data changes to be made available downstream with low latency.
The sample data used in this demo includes personally identifiable information (PII).
We’ll use AWS Glue, a fully managed ETL service, to stream the data from
Kinesis Data Streams, mask the PII found in our change data capture,
and store the transformed data in an Amazon S3 object store.
Let's begin by navigating to Amazon Relational Database Service (Amazon RDS),
where we’ll create our Amazon Aurora MySQL-compatible database.
First, we'll modify a few parameters specific to our database engine (DB engine)
for AWS DMS to get the CDC information.
Each DB engine has a specific set of parameter changes needed.
To learn what these are, please refer to the AWS DMS documentation for the
appropriate source DB engine.
Let’s save our changes.
Now we’ll create our database.
We’ll use the standard create method and specify the engine type as Aurora (MySQL Compatible).
Let’s use the latest available version of the engine.
We’ll use the Production template.
Let’s name the DB cluster “aurora-mysql-database-1.”
We’ll retain the username of “admin” and enter a password.
Let’s retain the default settings for the instance configuration and for availability and durability.
We’ll set up a connection to an Amazon Elastic Compute Cloud (Amazon EC2)
instance for this database.
For the purposes of this example, we have an EC2 instance already prepared.
We’ll retain the default subnet group settings.
The cluster is not publicly accessible.
We’ll use the default security group.
In this case, let’s make sure we’re using 3306 as our database port.
Let’s make some additional configurations.
We’ll specify an initial database name.
Next, we’ll select the DB cluster parameter group we created earlier.
Finally, let’s create the database.
Now that the database is available, let’s go to the EC2 instance that we’ll be using to interact with it.
For the purposes of this example, we’ve already created some files for interacting with this cluster.
The first one we’ll use creates a table called users.
The users table will include product- platform information such as the user
ID, username, first and last names, address, email, phone number, and the
number of documents modified, which is something we’re going to track.
We will then load a sample data set into the newly created users table.
Let’s go back to RDS and copy the writer instance endpoint for this database.
We’ll paste the endpoint into the MySQL command line and start loading the data.
We have 49,990 rows in this table.
Next, let’s navigate to Amazon Kinesis and create a data stream.
Let’s specify a name for our data stream.
Kinesis Data Streams provides On-demand and Provisioned capacity modes.
With on-demand capacity mode, there is no need for capacity planning--
Kinesis Data Streams instantly accommodates your workloads as they ramp up or down.
In this example, we are using the Provisioned capacity mode.
Let’s retain the default configuration and create the data stream.
Next, we’ll navigate to AWS Identity and Access Management (IAM) to create a
role that will be used by AWS DMS to interact with the data stream we just created.
The trusted entity will be an AWS service, namely DMS.
Let’s create a new policy to be used by this role.
We’ll specify Kinesis as the service.
In the actions, we’ll need three permissions.
The first one is DescribeStream, which is in the Read access level.
The other two are PutRecord and PutRecords, both of which are in the Write access level.
Next, we’ll specify a stream resource ARN.
Let’s specify the region we’re currently working in and the name of the stream we just created.
Let’s proceed.
We’ll specify a name and create the policy.
Now that we’ve created the policy, let’s go back and add it to the IAM role we’re creating.
Let’s refresh the view, select the policy, and proceed with the creation process.
Now that the role has been created, let’s take a closer look at it.
We’ll need the role’s ARN later on, so let’s copy it.
Next, we’ll navigate to AWS Database Migration Service to create a replication
instance, which will be used to run our migration tasks.
Let’s name the replication instance.
We’ll retain the default settings for the instance class and engine version.
We’ll select the failover option with multiple Availability Zones.
For the virtual private cloud, we’ll select the VPC our replication instances will run on.
We won’t make this instance publicly accessible.
Under Advanced settings, we can select an appropriate security group to connect
to the Aurora MySQL compatible database.
We’ll use the Default security group.
Let’s create the replication instance.
Next, we’ll create the endpoints that will allow AWS Database Migration
Service to read from and write to our database.
We’ll create the source endpoint first.
Let’s select an RDS DB instance.
We’ll select aurora-mysql-database-1- instance-1, which is our reader instance.
This will prepopulate our endpoint configuration.
Amazon Aurora MySQL is selected as the source engine, as expected.
Since we won’t be storing credentials with AWS Secrets Manager, we’ll
provide the access information manually.
Let’s enter the password.
Optionally, we can test the endpoint connection.
Let’s do so.
Let’s select the VPC designated for replication instances.
Let’s run the test.
The test was successful, which indicates that our source endpoint is able to connect
to the database using the specified replication instance.
Let’s create the endpoint.
Next, we’ll create the target endpoint.
We’ll specify Amazon Kinesis as the target engine.
For the service access role ARN, we’ll paste in the ARN of the IAM role we copied earlier.
Next, we’ll need the ARN for the Kinesis data stream we created.
Let’s navigate to Amazon Kinesis to copy it.
Now we’ll paste the ARN into our endpoint configuration.
We’ll leave the message format set to JSON.
Let’s run a test like we did with the source endpoint.
The test was successful, so let’s create the target endpoint.
Now that we have our two endpoints and our replication instance, we’ll
create the database migration task.
Let’s give the task an identifier and specify the replication instance we created earlier.
We’ll specify the two endpoints we created.
We’re going to track the number of documents that different users back up into the database.
For the migration type, we’ll select the option to migrate existing data
and replicate ongoing changes.
Let’s retain the default task settings.
In the table mappings, let’s add a new selection rule and enter a schema.
Next, we’ll specify the names of the source and source table we want to include.
We’ll leave the rest of the settings as they are and create the task.
The task has been created and is now running.
Let’s wait for the load to complete.
The load is now complete, and the task is performing ongoing replication.
Let’s drill down to the table statistics to see how long the load process took.
The elapsed load time is 4 minutes and 37 seconds.
As expected, the number of rows in the table matches what we saw earlier in the EC2 instance.
Let’s return there now.
We’ll use the Python SDK for AWS (Boto3) to read from the Kinesis data streams.
Let’s read and print the first 10 records.
This nested JSON structure includes a sequence number, a data element, and a metadata element.
Let's take a closer look at one of these records.
Notice the sequence number.
The data includes the raw, full row from the MySQL database.
The metadata information includes the record type, which is “data.”
The operation is “load.”
Next, we’ll perform some updates to the Aurora MySQL database and see
if they’re replicated to Kinesis.
Let’s create a duplicate window for the EC2 instance.
We’ll use one window to perform the updates and the other tab to perform the reads.
We’ll use this window for the reads.
We’ll connect to the database we created and perform a single row update.
At the same time, in this other window, we’ll perform a continuous read so that
Kinesis will look for any changes made from upstream.
Right now, this window does not show any changes because it’s reading from our
latest position in the Kinesis data stream.
Let’s go ahead and run an update.
User ID 25750 has increased the document count from 30417 to 30418.
Let’s take a look at how this change appears in the Kinesis data stream,
which is replicated via DMS.
The read found the update and provided the entire row.
Next, let’s take this a little further and perform continuous updates in the
upstream Aurora database.
This script will perform 20 updates, one after the other, in two-second intervals.
These changes will instantly be visible downstream in our Kinesis data stream.
As these updates are happening, we should see the changes being
replicated to Kinesis data stream instantly.
As expected, the changes are being replicated to the data stream.
This capability unlocks a lot of potential for downstream analytics use cases.
However, the data being passed on may contain a lot of Personally Identifiable Information (PII).
In a moment, we’ll see how to mask PII.
The updates have finished.
Next, we’ll use AWS Glue, our serverless extract, transform, and load (ETL) service,
to read the data from the Kinesis data stream, perform some updates that will
flatten the entire JSON payload, and mask some of PII such as email addresses and phone numbers.
In the left navigation panel, we’ll select Visual ETL.
Let’s create a job starting with a blank canvas.
Let’s give the job a name.
We’ll specify Amazon Kinesis as the source.
Next, we’ll select the stream name.
We’ll fetch everything from the earliest known position.
Let’s set the window size to 60 seconds.
Next, we’ll add some transformation actions.
Let’s flatten the entire JSON structure.
Since we want the data to be completely flattened out,
we’ll retain the default value of zero.
Next, we’ll detect sensitive data so we can mask it.
Let’s find columns that contain sensitive data.
Let’s retain the default settings and sample 100 percent of the data set,
with a detection threshold of 10 percent.
We already know that our data contains email addresses and US phone numbers,
so let’s browse for those patterns and designate them as sensitive information.
Let’s redact the selected PII text and replace it with a hash value.
Finally, we’ll store the results in an Amazon S3 bucket.
Let’s store the results in the Parquet format with a Snappy compression type.
We’ll specify a bucket that was created for this example.
We’ll use the data-output folder.
To store what we detect, let’s create a table in the AWS Data Catalog and
update the schema on subsequent runs.
We’ll use the default database.
Let’s enter a table name and save the job.
Let’s take a look at the script that has been generated based on the
transformations we’ve selected.
Let’s see what job details we still need to provide.
We need to specify an IAM role, so let’s do that now.
For our purposes, we’ve already created an IAM role with the required permissions
to read from Kinesis Data Streams and write to S3, along with the required Glue service permissions.
We’ll keep the other values as default and save this job.
Next, let’s run the job to start executing the ETL process.
Let’s take a look at the run status.
Right now, the job is running.
In a moment, this console will show real-time logs being ingested.
Logs are beginning to populate.
While the job is running, let’s navigate to the Glue Catalog and check our table list.
There’s a new table here.
Let’s take a quick look at the schema.
Next, let’s view the data in Amazon Athena.
We’ll query the Kinesis users table in the AWS Data Catalog.
Notice that the email addresses and phone numbers are masked.
Finally, let’s navigate to Amazon S3 and view the raw Parquet files in the data-output folder.
The data is partitioned by time, with folders based on the year, month, day, and hour.
Here are the raw Parquet files.
Let’s go back to the Glue console and stop the job run so we don’t get
charged for anything we don’t need.
To finish up, we’ll also want to delete the Kinesis data stream to avoid incurring costs.
You’ve just seen how to send CDC information from relational databases
to Amazon Kinesis Data Streams.
You can learn more about this topic in the description and links for this video.
Thanks for watching. Now it’s your turn to try.