\n
Amazon Kinesis Data Analytics for Apache Flink application
\n
In this notebook, a new Amazon Kinesis Data Analytics for SQL Application will be initialized and created. The notebook identifies the source/destination MSK topics and defines schema for the respective topics via AWS Glue Schema Registry. In addition, this notebook creates a streaming SQL query to aggregate data from the source MSK topic and store the results in the destination MSK topic.
\n
With a simple click of a few buttons the process to identify records in the source MSK topic, define logic and push it to the destination topic is accomplished. By using Flink SQL, the logic is understandable in natural language without any need to learn programming languages. Once the data transformation logic is defined in this notebook, you can build your code and export it to Amazon S3. You can promote the code that you wrote in your note to a continuously running stream processing application. After you deploy a note to run in streaming mode, Kinesis Data Analytics creates an application for you that runs continuously, reads data from your sources, writes to your destinations, maintains long-running application state, and autoscales automatically based on the throughput of your source streams. Documentation on this topic is available at https://docs.aws.amazon.com/kinesisanalytics/latest/dev/what-is.html.
\n
PLEASE NOTE WE DO NOT EXCECUTE INDIVIDUAL CELLS IN THIS NOTEBOOK, RATHER WE BUILD AND DEPLOY INTO AN EXECUTABLE APPLICATION.
\n\n
\n
Obtain MSK (Kafka) Cluster’s Connection String
\n
Before proceeding, fetch the MSK cluster connection string using below steps -
\n
\n - Navigate to MSK in AWS console.
\n - On the left hamburger menu, choose “Clusters”.
\n - Find the cluster pre-created by the CloudFormation template.
\n - Click on the cluster name.
\n - Hit button “View client information”.
\n - Under “Bootstrap servers” section, copy the “Plaintext” private endpoint and use that in topic create commands below.
\n - Replace the obtained connection string for property ‘properties.bootstrap.servers’. Just an example is shown below - DO NOT KEEP THE EXAMPLE STRING WHILE EXECUTING
\n
\n
\n
Define schema for source and destination topics
\n
We are using Flink’s Connector for Apache Kafka to interact with Kafka topics. The Kafka connector “maps” Kafka topics to a created table. When you are selecting the data from the table, the connector acts as topic consumer and reads the data from Kafka topic, when you are inserting data to the table, connector acts as Kafka producer and produces the messages to Kafka topic, the topic is denoted by ‘topic’ metadata provided in create table sql statement. See https://nightlies.apache.org/flink/flink-docs-release-1.13/docs/connectors/table/kafka/#how-to-create-a-kafka-table.
\n
When you create a table in a Kinesis Data Analytics Studio Notebook, an accompanying table in the Glue Data Catalog is also created. AWS Glue Data Catalog is a persistent metadata store that serves as a central repository containing table definitions. You can use the AWS Glue Data Catalog to quickly discover and search across multiple AWS datasets. Kinesis Data Analytics Studio is compatible with the AWS Glue Data Catalog, where you can define the schema for your source and destination tables. The reason for this is so that other applications, be it Apache Flink applications or Batch applications, etc, can reference the same schema that we defined in this notebook instead of creating an entirely different copy of the schema. This ensures consistency between applications working with the same data.
\n\n