Build a Real-Time Streaming Analytics Application on Apache Kafka

What We Will Accomplish

• Start a Serverless Amazon MSK Cluster
• Produce streaming data to MSK Serverless using Kafka Client Container
• Consume and process the streaming data using Amazon Managed Service for Apache Flink (MFA)
• Visualise streaming data in Amazon OpenSearch Service

Architecture

Implementation

• Java JDK
• Apache Maven

Step 1: Get the Repository

Step 2: Build the Flink Application

Step 3: Upload the File to Amazon S3

1. Log into your AWS account, navigate to the Amazon S3 console, and click Create bucket.

1. Provide a unique bucket name of your choice and choose an AWS region (e.g. us-east-1) and click Create Bucket at the bottom of the page. Take note of your bucket name.

1. Click on the newly created bucket and click Upload to upload the following file to the S3 bucket.

1. Click Add files and select the JAR file ClickStreamProcessor-1.0.jar that you have recently generated.

Step 4: Create a Stack using AWS CloudFormation

• Amazon OpenSearch Cluster: This is where we can visualize the consumed clickstream data. It is deployed in private subnets of a VPC.
• Amazon ECS Cluster + Task definition: The container application that generates the sample clickstream data runs inside the ECS cluster as a Fargate task.
• Amazon Managed Service for Apache Flink: This is where the Flink application runs, consuming the clickstream data from the MSK cluster, processing it and writing it to the OpenSearch Service.
• Amazon EC2 Instance (Kafka client): This EC2 instance serves as a Kafka client and allows us to interact with the MSK cluster by among other things creating Kafka topics.
• Amazon EC2 Instance (Nginx proxy): This EC2 instance serves as a Nginx proxy and allows us to access the OpenSearch Dashboard from outside of the VPC, i.e., from the Internet.
• Security groups: Security groups help us to control the traffic that is allowed to reach and leave a particular resource.
• IAM roles: An IAM role is an IAM identity that has specific permissions attached to it and can be assumed by an IAM user or an AWS service. For example, an IAM role can be used to grant permissions to an application running on an EC2 instance that requires access to a specific Amazon S3 bucket.

1. Navigate to the CloudFormation console and click on Create Stack.
2. Choose Upload a template file and click Choose file to upload the CloudFormation template file cf_template.yml that can be found in the root directory of the downloaded repository. Then, click Next.

1. Provide the stack with a Stack name of your choice (e.g. msk-serverless-stack). Additionally, you have to provide a value to the parameter AssetsBucketName. Enter the name of the S3 bucket that you created earlier. You can leave the default ClickstreamProcessor-1.0.jar as KdaAppKey unless you have changed the name of the JAR file that you have generated earlier. Leave the LatestAmiId as well as the OpenSearchmasterUserName as is. Click Next.

1. Scroll down the page Configure stack options and click Next.
2. Scroll down the page Review <Your_Stack_Name>. Make sure to tick the box that reads, I acknowledge that AWS CloudFormation might create IAM resources with custom names. Lastly, click Submit to create the CloudFormation stack.

Step 5: Create the MSK Serverless Cluster

1. Navigate to the Amazon MSK console and click Create cluster.
2. Choose Custom create and provide a cluster name of your choice (e.g. msk-cluster). Select Serverless as cluster type. Then, click Next.

1. In the Networking view, select the custom VPC named MMVPC. Then, click Add subnet to add a third subnet and choose the three available private subnets (PrivateSubnetMSKOne, PrivateSubnetMSKTwo, PrivateSubnetMSKThree) for the the different zones in us-east-1a, us-east-1b and us-east-1c.
2. Rather than the default security group, select the security group named MSK Security Group. Lastly, click Next.

1. Click Next.
2. Click Next.
3. Finally, click Create cluster to create the MSK Serverless cluster.
4. Once your MSK Serverless cluster status changes to Active, click on View client information.

1. Take note of the endpoint of your MSK Serverless cluster. Notice that we use IAM access control to handle the authentication to the MSK cluster.

Step 6: Create the Kafka Topics

1. Navigate to the Amazon EC2 console. On the EC2 home page click on Instances (running).

1. On the EC2 Instances page select the checkbox for the instance named KafkaClientInstance and click on the Connect button on top right as shown in the image below.

1. On the page Connect to instance, ensure to select Session Manager and click the Connect button. This opens a new tab with an EC2 terminal window.
2. In the terminal window execute the following command to change to ec2-user:

1. Execute the command below to set your MSK cluster endpoint to the shell variable BS. Please replace <Your_Cluster_Endpoint> with the endpoint you noted down after you created the MSK Serverless cluster.

1. Then, execute the following command to create the Kafka topic.

1. You should see a single MSK topic that has been created: clickstream.

Step 7: Start a Container Application to Generate Clickstream Data

1. Navigate to the Amazon ECS console. On the left side menu click on Task Definitions to view all available Task definitions. Select the checkbox of the available Task definition and select Run task from the Deploy menu.

1. On the Run Task page, select the existing cluster (msk-serverless-[...]-cluster) and leave the default settings.

1. Expand the Networking section. Change the default VPC to MMVPC. Similar to before, select the three subnets PrivateSubnetMSKOne, PrivateSubnetMSKTwo and PrivateSubnetMSKThree. Finally, uncheck the default security group and select the security group that contains -ProducerECSTaskSecurityGroup-.

1. Expand the Container overrides section. For BOOTSTRAP_STRING enter the value to your MSK Serverless cluster endpoint (written down earlier with View client information from the MSK cluster console page).

1. Finally, click the Create button.
2. Wait for your task to change to the Running status as shown below.

Step 8: Check Schema in AWS Glue Schema Registry

1. Navigate to the Amazon Glue console. Select Stream schema registries under Data Catalog from the left menu. You can see the schema registry named serverless. Click on it.
2. You can see the available schemas of the schema registry serverless. Click on the schema clickstream to see the different schema versions. You should see version 1 here.

1. Click on the version 1 to see the Avro schema of the clickstream data produced by the ECS task.

Step 9: Consume Clickstream Data Using Managed Service for Apache Flink

1. Navigate to the Managed Apache Flink console and click on the open streaming application KDAFlinkCLickstream-msk-serverless-stack.
2. Configure and update the application by clicking on the Configure button.

1. Scroll down to the Runtime properties. Update BootstrapServers to the MSK Serverless cluster endpoint you have written down earlier. Keep the rest of the values as default.

1. Finally, save your changes.
2. Click on the Run button to run the Flink application. Choose Run without snapshot.

1. Once the Flink application is running, click on Open Apache Flink dashboard to open the Flink dashboard.

1. Click on Running Jobs on the left side of the menu. Click on Flink Streaming Job to access the details of the running job.

1. This opens a screen with a directed acyclic graph (DAG), representing the flow of data throughout each of the operators of your application. Each blue box in the job workflow represents a series of chained operators, known as Tasks in Apache Flink.

Step 10: View Clickstream Data in the Amazon OpenSearch Dashboard

1. Navigate to the CloudFormation console and click on the stack that we created earlier. Go to the Outputs tab of the stack.
2. Take note of the OpenSearchMasterUserName and OpenSearchMasterPassword. We will need the values in the next step.

1. Click on the OpenSearchDashboardEndpoint to open the OpenSearch Dashboard login page in a new tab. As the OpenSearch Service is deployed in a VPC, we are relying on Nginx reverse proxy to access the OpenSearch Dashboard outside of the VPC. Note that we are using a self signed certificate for Nginx. However, we strongly recommend using a valid certificate for production deployments.
2. If you are accessing the URL using Google Chrome, click on the Advanced button and click on Proceed to <Your_EC2_DNS>.

1. Use the OpenSearchMasterUserName and OpenSearchMasterPassword from the previous step and log into to OpenSearch Dashboards.

1. Select Global tenant on the popup dialog box.
2. Click on the hamburger menu on the left of the screen and click on Dashboards as shown below.

1. In the Dashboards view select the dashboard named Clickstream Dashboard to see the plotted data:

Clean Up the Resources

• Delete the MSK Serverless Cluster under the Actions menu.

• Delete the CloudFormation Stack.

• Empty and delete the S3 Bucket you have created earlier.

Conclusion

• Data processing with Kafka Streams - An overview of stateless operations
• In the land of the sizing, the one-partition Kafka topic is king
• My Event is More Urgent than Yours: Prioritizing Event Processing with Apache Kafka