In this video, you’ll see how to join data from Change Data Capture (CDC) and
streaming sources to enhance analytics.
Using Amazon Kinesis Data Analytics for Apache Flink, you can create a
Studio notebook for interactive data analytics, query CDC data from a
relational database, and use JOIN queries to enrich CDC data with streaming data.
This table illustrates the performance and complexity trade-offs between
several streaming data enrichment patterns in Apache Flink.
For this video, we’ll demonstrate joins using the Table/SQL API pattern.
We’ll be using Amazon Kinesis Data Analytics for Apache Flink to read and
join data from two sources:
an AmazonAurora MySQL relational database and an Amazon Kinesis Data Stream.
A CDC connector allows Apache Flink to ingest the initial data from Amazon
Aurora MySQL, and thereafter capture only the changed records.
Apache Flink also ingests events from an Amazon Kinesis Data Stream.
Enriched data resulting from transformations or aggregation within Amazon Kinesis Data
Analytics for Apache Flink is then sent to Amazon Simple Storage
Service (Amazon S3) for storage and use in further analytics.
Let’s start by navigating to Amazon Kinesis Data Analytics Studio.
Studio notebook allows us to query data streams in real-time using SQL,
Python or Scala, to build a stream processing application.
Let’s create a new Studio notebook.
We’ll choose to create our notebook with custom settings.
Let's give our application a name.
We’ll leave the default Apache Flink runtime selected.
The notebook will create an AWS Identity and Access Management (IAM) role for us.
We'll need to specify an AWS Glue database that defines metadata for
our sources and destinations.
We’ll select the default.
Let’s proceed to the next step.
We’ll retain the default Scaling and Logging and monitoring settings
and move our focus to Networking.
The first option, Custom VPC configuration, requires a private routable network path from
the Studio notebook to the database VPC.
To demo what it would look like if your source of streaming was Amazon MSK,
we’ll select VPC configuration based on Amazon MSK cluster, which places the
Studio notebook in the same VPC as the Amazon MSK and source databases.
Let’s specify the Amazon MSK cluster.
The cluster we selected has three subnets.
Several AWS-supported connectors are added by default out of the box.
In addition, we’ll be adding a custom connector for capturing CDC data from
our Amazon Aurora MySQL database.
This GitHub page provides a variety of CDC Connectors for Apache Flink.
You can access the connectors and their documentation here.
For the purposes of this demonstration, we have already downloaded the mysql-cdc
connector and placed it in an Amazon S3 bucket.
Here’s the connector.
Let’s copy the bucket’s S3 URI so we can add this connector to our Studio notebook.
Now, we’ll add our custom connector.
First, we’ll choose the Amazon S3 bucket of our connector.
We’ll paste in the path to the S3 object.
We can remove the bucket information, which is already specified in the box above.
Let’s save these changes.
The custom connector has been added to the notebook configurations.
Let’s continue.
Here we’ll quickly review our chosen settings and create the Studio notebook.
Our Studio notebook has been created.
It takes a few minutes to create the underlying Apache Flink cluster.
For our purposes, we’ll proceed with an existing Studio notebook that has the same settings.
Let’s open the notebook in Apache Zeppelin.
The notebook is already populated with the code we need to interact with our source data.
In the first code block, we’re enabling once in a minute checkpointing for the application.
In the next code block, we’ll instantiate a MySQL client using the mysql-connector-python
library in order to interact with our database from the notebook.
The next two statements will be used later to insert a record in our
source database to verify CDC.
The next code block has SQL statements we can use to create tables that correspond
to our two different data sources.
The customer table connects to the MySQL database using the mysql-cdc connector.
The order table corresponds to our Kinesis data stream using the
built-in kinesis connector for Flink.
The table columns correspond to the schema of each event in our stream.
Our new tables have been created and have been added to our AWS Glue Data Catalog.
We can now use select queries to read from our data sources.
First, we’ll verify the CDC data collector by reading from our MySQL customer table.
Let’s filter for customer IDs greater than 1998.
For each query, the Studio notebook submits a job to the Apache Flink cluster,
which then emits the results to our notebook.
We have selected customers with an ID greater than 1998.
Let’s insert a record with an ID of 2002 and see what happens.
The new record immediately appears in the results.
The Flink job is ingesting all the CDC data and constantly reading
the underlying database changes.
Let’s see what happens when we update a record.
Let’s update the record that we just inserted with by changing the name
from “Customer 2002” to “Customer Name 2002.”
As expected, the results were updated.
Next, let’s verify the kinesis connector.
To do this, we’ll select from the order table we created, which reads from our
Amazon Kinesis Data Steam.
Each data event has a customer ID, item name, order amount, and an order time.
Every time we run this job, we should see all the data events that exist in this stream.
The next block of code enriches these data events with the customer information
so we can run real-time analytics to see the total order amount per market segment.
In this query, the select statement joins the customer table with the order table on
the customer_id field, even though the tables are coming from different sources.
We also calculate the summation of order amounts per customer.
Our results group the totals by customer ID, customer name, and market segment.
To see the sum of the order amounts per market segment, we can switch
from the table view to a bar chart with market segment as the key field.
We’re can now see order totals for each market segment.
Presently, our stream producer isn’t running,
so we don't see any changes from events in the data stream.
Now let’s start our stream producer.
As data is received by the Kinesis Data Stream, it becomes available to Apache Flink.
As more data streams in, the results of the calculation update in real time.
You can now add code to send these results to Amazon S3 using
the Flink FileSystem connector.
You’ve just seen how to join data from CDC and streaming sources to enhance analytics.
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.