Suppose we have a Producer microservice and a Consumer microservice, whenever a new addition happens to the Producer microservices it should be reflected with the Consumer microservice in real-time.

So there should be a way of capturing the changes done to the source data system and propagating them to target data systems in a reliable and scalable manner in real-time.

And this paves the road for Change Data Capture.

This article will discuss how to capture the changes from the source database and propagate them to the target database using MySQL, Debezium Server, and Amazon Kinesis.

Understanding CDC, MySQL Binlogs, Debezium Server and Amazon Kinesis

So before we jump into the implementation of this system we will need to understand a few concepts.

Change Data Capture

Change Data Capture (CDC) tracks data changes (usually close to real-time). CDC can be implemented for various tasks such as auditing, copying data to another system, or processing (and reacting to) events. In MySQL, the easiest and probably most efficient way to track data changes is to use binary logs.

MySQL Binlogs

The binary log is a set of log files that contain information about data modifications made to a MySQL server instance.

Debezium

• Debezium is a set of distributed services to capture changes in our databases so that the applications can see those changes and respond to them
• Debezium records all row-level changes within each database table in a change event stream, and applications simply read these streams to see the change events in the same order in which they occurred

Debezium Server

• The Debezium server is a configurable, ready-to-use application that streams change events from a source database to a variety of messaging infrastructures
• Change events can be serialized to different formats like JSON or Apache Avro and then will be sent to one of a variety of messaging infrastructures such as Amazon Kinesis, Google Cloud Pub/Sub, or Apache Pulsar
• For this POC we will be using Amazon Kinesis as the data streaming means

Amazon Kinesis

• Amazon Kinesis Data Streams (KDS) is a massively scalable and durable real-time data streaming service
• We can create data-processing applications, known as Kinesis Data Streams applications

Configuration Steps

Let’s dive into the tech stack configurations that is needed for this POC. They are as below:

1. MySQL

Before getting started with CDC in MySQL, we need to enable binary logging. Steps for enabling binary logging :

Step 1: Checking the Database Status

The first step is to check if the log-bin option is turned on or not and what binary log format is used. MySQL provides three types of bin-log formats: ROW, MIXED, and STATEMENT.

SHOW VARIABLES LIKE 'log_bin'; -- Value should be ON
SHOW VARIABLES like '%binlog_format%'; -- Value should be set to ROW

Step 2: Updating the MySQL Configuration File

If the log-bin value is OFF from the above query, we need to update the MySQL configuration file (my.conf) by adding the below lines to the [mysqld] section.

log-bin=bin.log
log-bin-index=bin-log.index
max_binlog_size=100M
binlog_format=row
socket=mysql.sock

Once Step 2 is completed, confirm the changes using the queries in Step 1.

With the above steps, we are all set with the MySQL configurations for CDC.

2. Installing and Configuring Debezium Server

Step 1. Install Java

Make sure that you have Java installed, if not install Java.

apt-get install openjdk-8-jre

Step 2. Install Debezium Server

• For making this POC we will be using the Debezium server version 1.5.1.Final, download the version using the below command:

wget -O debezium.tar.gz
"https://repo1.maven.org/maven2/io/debezium/debezium-server-dist/1.5.1.Final/debezium-server-dist-1.5.1.Final.tar.gz"

• Unzip the downloaded content to the disk

tar -xf debezium.tar.gz 

• A directory named debezium-server will be created with the following contents:

Step 3. Configuring the Debezium Server

• Create a config file called application.properties in the folder conf/

touch conf/application.properties

• Create a new folder called data and add a new file called offsets.dat

mkdir data
touch data/offsets.dat

• The new folder structure will be as shown below

• Now in the application.properties file that we created in the previous step, add the below configurations:

debezium.sink.type=kinesis
debezium.sink.kinesis.region=us-west-2
debezium.sink.kinesis.credentials.profile=default
debezium.source.connector.class=io.debezium.connector.mysql.MySqlConnector
debezium.source.offset.storage.file.filename=data/offsets.dat
debezium.source.offset.flush.interval.ms=0
debezium.source.database.hostname=localhost
debezium.source.database.port=3306
debezium.source.database.user=root
debezium.source.database.password=root
debezium.source.database.server.name=debezium-tutorial
debezium.source.schema.include.list=TestDB
debezium.source.table.include.list=TestDB.Orders

debezium.source.database.history=io.debezium.relational.history.FileDatabaseHistory
debezium.source.database.history.file.filename=history.dat

Note:

1. debezium.sink.type: is for the type of messaging or streaming partner that we use to stream the data. For this POC we are using Amazon Kinesis.
2. debezium.sink.kinesis.region: is the region to which the Kinesis stream is configured with.
3. debezium.sink.kinesis.credentials.profile: is to mention the profile details for which the Kinesis stream is created (only used when we have multiple AWS profiles configured in our system, else it will be the default).
4. debezium.source.connector.class: specifies the type of database management system from which the changes are captured. For this POC we are using a MySQL connector. Connectors for other database management systems are available.
5. debezium.source.offset.storage.file.filename: is the file in which the connector offsets are stored.
6. debezium.source.database.dbname: is the name of the database whose changes are to be captured.
7. debezium.source.database.server.name: is the name of the running debezium server, mainly used while creating streams in Amazon Kinesis.
8. debezium.source.table.include.list: this is the list of table names whose changes are to be captured separated by commas. If this configuration is not provided we need to create streams for all individual tables present within the Database for which the changes are to be captured. The names of the tables should be provided as database_name.table_name . In order to capture the changes from multiple tables the names of the tables have to be provided by separating with commas. For example database_name.table_name1, database_name.table_name2, ......
9. debezium.snapshot.new.tables = parallel: this configuration is used to take snapshots of the newly created tables after the debezium server is configured. If this is not provided then that tables’ changes won’t be captured/snapshotted.
10. debezium.source.tombstones.on.delete=false: This configuration is mainly used to capture the changes made by DELETE queries. Usually, a delete operation is represented by a delete event and a subsequent tombstone event, which causes the server to exit unexpectedly.
11. debezium.source.database.history: Some of the connectors (e.g MySQL, SQL Server, Db2, Oracle) monitor the database schema evolution over time and store the data in database history. io.debezium.relational.history.KafkaDatabaseHistory is the default value for Kafka bases systems. io.debezium.relational.history.FileDatabaseHistory is used for non-Kafka deployments.
12. debezium.source.database.history.file.filename: The name and location of the file to which FileDatabaseHistory persists its data.

3. Setting up Amazon Kinesis

Step 1. Creating Kinesis Data Streams

We need to create Data Streams for two purposes, they are

1. Stream to identify the debezium server which sent the messages, in the same name as provided in debezium.source.database.server.name
2. Streams to capture the changes for the individual tables, ie: we have to create Kinesis streams for all the tables listed in debezium.source.table.include.list , the format for creating the streams are:

<debezium.source.database.server.name>        -- Stream for Server
<debezium.source.database.dbname>.<tablename> -- Stream for tables

for example, based on our application.properties file

debezium-tutorial                             -- Stream for Server           
debezium-tutorial.TestDB.Events               -- Stream for table

Steps to create a Kinesis Data Stream

1. Go to the Amazon Kinesis page and click Create data stream
2. Enter the data stream name
3. Add the required number of shards, for out POC 1 shard is enough
4. Click create data stream

Similarly, create the necessary streams for the tables provided in debezium.source.table.include.list

Step 2. Lambda function to read the streaming data

Once the data reaches the Kinesis streams we can process it and can route it to the respective target points.

Here we will be writing a lambda function that will get triggered based on incoming data streams.

Steps to create the lambda function:

1. Go to the AWS Lambda page and click Create function
2. Select “Auther from scratch” and provide the function name and select the Runtime as Python 3.8
3. Click Create function

Configuring the roles created

1. Once lambda function is created go to the configurations tab on the lambda function page
2. Choose the Permission tab where we can see the Role that we have created

3. Click on the Role name, this will open up the role’s summary page

4. Click attach policies and add “AmazonKinesisFullAccess”.

Add Triggers

1. Go to the configurations tab on the lambda function page
2. Choose the Triggers tab where we can add new triggers
3. Select Kinesis from the dropdown and add the streams created by us earlier from the Kinesis streams dropdown

Add code to Lambda Function

lambda_function.py

import base64
import json
import boto3

# simple lambda function to retrieve the data from Kinesis Data Stream
# in order to send the data to the respective target DB's ..
# make the necessary connections

def lambda_handler(event, context):
    kinesis = boto3.client('kinesis')
    for record in event['Records']:
        payload = base64.b64decode(record['kinesis']['data']).decode('utf-8')
        data = json.loads(payload)
        # do something with the data
        print(data)
    return 'Successfully processed {} records.'.format(len(event['Records']))

Now we are ready to go with the testing…

Testing the CDC Pipeline

1. Create a database called TestDB

Create database TestDB;

2. Create a test table called Events

create table `TestDB`.`Events` (
  `Id` int not null,
  `EventName` varchar(150) null,
  `EventDate` datetime null,
  primary key(`Id`));

3. Start the Debezium server from the debezium-server folder using

sh run.sh

4. Once the server is up, we can start doing INSERT, UPDATE, DELETE operations to the table created :

Insert into Events values (1,'Test',CurDate());
Update Events set EventName = "updated test" where Id=1;
Delete from Events where Id=1;

We can check the data received at the Kinesis stream by checking the cloudwatch logs.

Similarly, we will be receiving data streams with op values as u and d respectively for UPDATE and DELETE operations.

The received data can then be routed to the required target databases based on our requirements.

References

• Debezium Server
• Debezium connector for MySQL
• io/debezium/debezium-server-dist/1.5.1.Final
• Real-Time Streaming Analytics - Amazon Kinesis Data Streams - AWS

Hope the content was informative! :)

Noufal Rijal

Noufal Rijal - Software Development Engineer - InApp | LinkedIn

The winter storms in Texas in 2021 have made our client come up with a use case that helps them collect data from users through handwritten paper form, and save the data points to our existing database tables.

As developers, this was a new challenge and an opportunity to learn and try new technology stacks.

After a good round of research, we finalized to use the OCR service provided by AWS called “Textract”.

When we were working on it we couldn’t find many resources which could guide us to work with these services, so I on behalf of our team thought of documenting and publishing our journey…

Amazon Textract is a machine learning service that automatically extracts text, handwriting, and data from scanned documents that goes beyond simple optical character recognition (OCR) to identify, understand, and extract data from forms and tables.

In this post we will be covering the following checkpoints:

1. How to upload image/document to S3 bucket

2. How to create a Lambda function that gets triggered when a new image or doc is uploaded in S3

3. How to write python code in lambda function which will help in parsing the analyzed text/data from doc/image and provide a JSON output

4. Creating a new response JSON file whenever a new image/doc is uploaded to s3

Prerequisites

1. You need to have an AWS account and some basic knowledge of AWS services.

2. To use AWS Textract in Python, the latest “boto3” package is needed which is not currently available in AWS Lambda hosted environments. Execute this command: pip install — target ./python boto3

3. You should have created essential credentials like Access Key and Secret Key on your account.

4. Following AWS services will be utilized throughout this guide

• Lambda Service
• Textract Service
• Simple Storage Service i.e. S3
• Identity Access Management Service

1. Uploading Image/doc to S3 bucket

Create S3 Bucket

1. Go to AWS S3 page and click Create Bucket

2. Enter the Bucket name and Region click Next

3. Set permissions and click Create Bucket

Note:- The region selected should be the same as the one used to create the Lambda function

Upload image to S3 bucket

For this example, we are trying to upload images directly from the AWS S3 bucket console.

2. Extracting data from an S3 image

This process consists of a Lambda function that gets triggered whenever an image (.jpg extension) gets uploaded to the S3 bucket.

Follow the steps below

Creating the S3 Lambda Trigger

Steps to create a Lambda function that gets triggered whenever an image is uploaded

1. Go to AWS Lambda page and click Create function

2. Select “Use a Blueprint” and search for “s3-get-object-python” template and click “configure”

3. Enter the function name, role name

4. For creating the S3 trigger under the S3 trigger tab:

• Add the bucket name
• Select all object to create events for Event type
• Suffix as .jpg

5. Click create function

Note:- We can add more than one triggers for PNG, PDF files based on our needs

Configuring the roles created

1. Once lambda function is created go to the configurations tab on the lambda function page

2. Choose the Permission tab where we can see the Role that we have created

3. Click on the Role name, this will open up the role’s summary page

4. Click attach policies and add “AmazonTextractFullAccess”, this will give the lambda function to access the AWS Textract service

Add Layers to Lambda function

Step 1- Creating the boto3 zip

1. In the local machine create a directory for the project

mkdir -p zip_boto3/python

2. Install boto3 package in to boto3/python directory

Pip install boto3

3. Zip the contents of boto3

cd zip_boto3

zip -r boto3-layer.zip python

Step 2- Creating layer in AWS

1. Log in to AWS console and go to lambda

2. Select layers and click create a layer

3. Fill in the details for the layer and select Python 3.7 as the compatible runtime

Step 3-Using the layer

1. Select the lambda function that we have created

2. Click on Layers and then Add a layer

3. Pick the layer that we have created and save the lambda function

3. Add code to Lambda Function

Create two files in your function folder

lambda_function.py

Note:- The above code snippet will help in capturing the table data from the uploaded file, and returns a JSON file with the table contents specified by the column name

trp.py

Save and test your Lambda function

4. Result

Whenever an object with .JPG extension is uploaded to the S3 Bucket the associated lambda function will get triggered and an output file with .JSON extension will be generated back to S3 Bucket.

5. References

• Automatically extract text and structured data from documents with Amazon Textract | Amazon Web Services
• Working with object metadata
• Textract - Boto3 Docs 1.20.25 documentation
• amazon-textract-response-parser
• Textract - Boto3 Docs 1.20.25 documentation

Hope the content was informative! :)

-Noufal Rijal