{"id":5761,"date":"2023-04-14T09:00:00","date_gmt":"2023-04-14T07:00:00","guid":{"rendered":"https:\/\/blog.besharp.it\/?p=5761"},"modified":"2023-04-13T16:29:58","modified_gmt":"2023-04-13T14:29:58","slug":"efficiently-stream-dml-events-on-aws","status":"publish","type":"post","link":"https:\/\/blog.besharp.it\/efficiently-stream-dml-events-on-aws\/","title":{"rendered":"Efficiently stream DML events on AWS"},"content":{"rendered":"\n
In the past, many workloads came with their own big and monolithic database, where not only the application but also reporting tools and technical support connected to it and performed queries. <\/p>\n\n\n\n
As this is still true today, companies are moving towards having single information stored on multiple data sources and servers. Only the core application should be able to access the database, reporting tools should use data that is stored on a separate instance, and monitoring and data analytics should be done by aggregating data that comes from different sources.<\/p>\n\n\n\n
To do this we need to stream the changes that are occurring on our database to one or more destinations. Today we are going to take a look at how to do this on AWS.<\/p>\n\n\n\n
AWS Database Migration Service (DMS)<\/strong> is a powerful tool for migrating data between various database platforms. One of the standout features of AWS DMS is its Change Data Capture (CDC) functionality, which allows for real-time streaming of changes made to a source database to a target database.<\/p>\n\n\n\n When using AWS DMS, you have the option to attach a target database directly as an endpoint or use Amazon Kinesis Data Streams to capture and process the streaming data. <\/p>\n\n\n\n Here are some differences between the two approaches:<\/p>\n\n\n\n Overall, both approaches have their advantages and disadvantages, and the best choice depends on your specific use case and requirements. In this article, we are going to explore the possibility to process insert\/update\/delete events on flight with the help of Amazon Kinesis Data Streams.<\/p>\n\n\n\n Now, let\u2019s build a proof of concept to test out the CDC streaming solution with DMS and Kinesis Data Streams. The idea is to have an automated process that gives us an easy way to replicate changes that happen on a source database to one or more destination engines.<\/p>\n\n\n\n This is a diagram of what we\u2019re going to build:<\/p>\n\n\n\n <\/p>\n\n\n <\/p>\n\n\n\n The first thing we need to do, if we want to enable CDC, is configure our source database<\/strong> to make all the information needed by DMS available to trap new events. For many engines, this means running a bunch of queries that are well-described in the official AWS documentation<\/a>. <\/p>\n\n\n\n After the source database has been configured, let\u2019s create our Kinesis Data Stream<\/strong>. <\/p>\n\n\n\n This step is pretty straightforward as we don\u2019t need to provide many parameters. We only need to decide what\u2019s our data stream\u2019s capacity mode: <\/p>\n\n\n\n In our case, we chose the on-demand capacity mode. Keep in mind the concept of sharding<\/strong>, it\u2019ll come back later on in this article.<\/p>\n\n\n\n Now that we have the source and target systems configured, we need to create the AWS DMS instance and endpoints,<\/strong> which are basically a set of configurations that DMS uses to interact with the various systems. <\/p>\n\n\n\n Configuring a DMS endpoint is really straightforward: first, you need to choose whether you are creating a source or target endpoint, then you specify the type of engine that the endpoint is connected to, and finally, depending on the selected engine, you need to provide a few parameters to configure the connection. <\/p>\n\n\n\n Here\u2019s an example of the creation form for the Kinesis Data Stream endpoint:<\/p>\n\n\n\n <\/p>\n\n\n <\/p>\n\n\n\n AWS DMS requires an IAM role to be able to interact with Kinesis. This role should have at least these permissions:<\/p>\n\n\n\n Now the last resource to create<\/strong> on the AWS DMS console is the actual task that will be taking care of copying data from our source to Kinesis<\/strong>. <\/p>\n\n\n\n Choose the DMS replication instance where you want to deploy your task, select the endpoints you previously created and configure the type of migration you want to perform: you can have a migration that takes a snapshot of the data that is stored on the source engine, or you can have change data capture enabled so that new events are automatically replicated to the target engine, or you can have both.<\/p>\n\n\n\n Replication tasks have tons of configuration parameters you can tweak to increase the efficiency, stability, and speed, but I\u2019m not going to deep dive into them; just know that, for this POC, I didn\u2019t change anything. Everything is set to its default value.<\/p>\n\n\n\n The task creation will take a few minutes, in the meantime we can focus on creating the resources for\u2026<\/p>\n\n\n\n On the architecture diagram, I put AWS Glue<\/strong> as the processing service because of its native integration with many data warehouses and for the possibility to ingest and process big datasets thanks to Spark. Furthermore, AWS Glue Jobs can be configured in streaming mode, meaning that they are always on and processes incoming data as it comes.<\/p>\n\n\n\n Here you can find a small code snippet that can help you kickstart a processing job:<\/p>\n\n\n\n <\/p>\n\n\n\n <\/p>\n\n\n\n Check the GitHub Gist<\/a> to copy the code.<\/p>\n\n\n\n This code basically creates a link between the Glue Job and a Data Frame that has Kinesis Data Stream as the source. This Data Frame will be refreshed every 100 seconds, and every new version of it will be passed to the process_batch<\/em> function. This function takes care of the processing of new data.<\/p>\n\n\n\n When creating the Kinesis Data Frame, we can provide a schema format for the incoming data<\/strong>. <\/p>\n\n\n\n This schema must match the format of the data that DMS provides us, which is a JSON that contains two keys:<\/p>\n\n\n\n As you can see, I only provided a \u201cstructured\u201d schema for the metadata part. This is because we can\u2019t possibly know the format of the \u201cdata\u201d key before knowing the name of the schema and table. <\/p>\n\n\n\n After starting our new Glue job, we should see some logs on CloudWatch similar to these:<\/p>\n\n\n\n <\/p>\n\n\n\n <\/p>\n\n\n\n Finally, I wanted to dump the events on Amazon S3<\/strong> as they come from DMS, so I can always decide to go back and reprocess everything up to a certain point in time. We already have our Data Stream on Kinesis, and we can attach multiple consumers to it, so we can easily create a Kinesis Firehose Delivery Stream to aggregate events and dump them on an S3 bucket.<\/p>\n\n\n\n When we tried for the first time this solution, we bumped into a performance problem related to the scalability of our Data Stream: as we discussed before, Kinesis uses a concept of shards as the unit to measure the scaling. When the producer produces a new event, it has to provide the shard where the event has to go or use a random ID to make Kinesis decide.<\/p>\n\n\n\n DMS always uses the same value.<\/p>\n\n\n\n By doing so, it always uses the same shard, so even if you configured your Data Stream\u2019s capacity mode as on-demand, it won\u2019t scale because of DMS.<\/p>\n\n\n\n\n
Setup efficient DML events streaming on AWS <\/h2>\n\n\n\n
<\/figure><\/div>\n\n\n
The ingestion<\/h3>\n\n\n\n
\n
\n
<\/figure><\/div>\n\n\n
{\n\t\"Version\": \"2012-10-17\",\n\t\"Statement\": [\n \t\t{\n \t\t \t\"Action\": [\n \t\t\t \t\"kinesis:PutRecord\",\n \t\t\t \t\"kinesis:PutRecords\",\n \t\t\t \t\"kinesis:DescribeStream\"\n \t\t\t],\n \t\t\t\"Resource\": [\n \t\t\t\t\"<your delivery stream arn>\"\n \t\t\t],\n\t\t \t\"Effect\": \"Allow\"\n \t\t}\n\t]\n}<\/code><\/pre>\n\n\n\n
The processing<\/h3>\n\n\n\n
<\/figure>\n\n\n\n
\n
<\/figure>\n\n\n\n
Possible issues<\/h3>\n\n\n\n