Build Spark Structured Streaming applications with the open source connector for Amazon Kinesis Data Streams

Apache Spark is a powerful big data engine used for large-scale data analytics. Its in-memory computing makes it great for iterative algorithms and interactive queries. You can use Apache Spark to process streaming data from a variety of streaming sources, including Amazon Kinesis Data Streams for use cases like clickstream analysis, fraud detection, and more. Kinesis Data Streams is a serverless streaming data service that makes it straightforward to capture, process, and store data streams at any scale.

With the new open source Amazon Kinesis Data Streams Connector for Spark Structured Streaming, you can use the newer Spark Data Sources API. It also supports enhanced fan-out for dedicated read throughput and faster stream processing. In this post, we deep dive into the internal details of the connector and show you how to use it to consume and produce records from and to Kinesis Data Streams using Amazon EMR.

Introducing the Kinesis Data Streams connector for Spark Structured Streaming

The Kinesis Data Streams connector for Spark Structured Streaming is an open source connector that supports both provisioned and On-Demand capacity modes offered by Kinesis Data Streams. The connector is built using the latest Spark Data Sources API V2, which uses Spark optimizations. Starting with Amazon EMR 7.1, the connector comes pre-packaged on Amazon EMR on Amazon EKS, Amazon EMR on Amazon EC2, and Amazon EMR Serverless, so you don’t need to build or download any packages. For using it with other Apache Spark platforms, the connector is available as a public JAR file that can be directly referred to while submitting a Spark Structured Streaming job. Additionally, you can download and build the connector from the GitHub repo.

Kinesis Data Streams supports two types of consumers: shared throughput and dedicated throughput. With shared throughput, 2 Mbps of read throughput per shard is shared across consumers. With dedicated throughput, also known as enhanced fan-out, 2 Mbps of read throughput per shard is dedicated to each consumer. This new connector supports both consumer types out of the box without any additional coding, providing you the flexibility to consume records from your streams based on your requirements. By default, this connector uses a shared throughput consumer, but you can configure it to use enhanced fan-out in the configuration properties.

You can also use the connector as a sink connector to produce records to a Kinesis data stream. The configuration parameters for using the connector as a source and sink differ—for more information, see Kinesis Source Configuration. The connector also supports multiple storage options, including Amazon DynamoDB, Amazon Simple Service for Storage (Amazon S3), and HDFS, to store checkpoints and provide continuity.

For scenarios where a Kinesis data stream is deployed in an AWS producer account and the Spark Structured Streaming application is in a different AWS consumer account, you can use the connector to do cross-account processing. This requires additional Identity and Access Management (IAM) trust policies to allow the Spark Structured Streaming application in the consumer account to assume the role in the producer account.

You should also consider reviewing the security configuration with your security teams based on your data security requirements.

How the connector works

Consuming records from Kinesis Data Streams using the connector involves multiple steps. The following architecture diagram shows the internal details of how the connector works. A Spark Structured Streaming application consumes records from a Kinesis data stream source and produces records to another Kinesis data stream.

A Kinesis data stream is composed of set of shards. A shard is a uniquely identified sequence of data records in a stream and provides a fixed unit of capacity. The total capacity of the stream is the sum of the capacity of all of its shards.

A Spark application consists of a driver and a set of executor processes. The Spark driver acts as a coordinator, and the tasks running in executors are responsible for producing and consuming records to and from shards.

The solution workflow includes the following steps:

1. Internally, by default, Structured Streaming queries are processed using a micro-batch processing engine, which processes data streams as a series of small batch jobs. At the beginning of a micro-batch run, the driver uses the Kinesis Data Streams ListShard API to determine the latest description of all available shards. The connector exposes a parameter (kinesis.describeShardInterval) to configure the interval between two successive ListShard API calls.
2. The driver then determines the starting position in each shard. If the application is a new job, the starting position of each shard is determined by kinesis.startingPosition. If it’s a restart of an existing job, it’s read from last record metadata checkpoint from storage (for this post, DynamoDB) and ignores kinesis.startingPosition.
3. Each shard is mapped to one task in an executor, which is responsible for reading data. The Spark application automatically creates an equal number of tasks based on the number of shards and distributes it across the executors.
4. The tasks in an executor use either polling mode (shared) or push mode (enhanced fan-out) to get data records from the starting position for a shard.
5. Spark tasks running in the executors write the processed data to the data sink. In this architecture, we use the Kinesis Data Streams sink to illustrate how the connector writes back to the stream. Executors can write to more than one Kinesis Data Streams output shard.
6. At the end of each task, the corresponding executor process saves the metadata (checkpoint) about the last record read for each shard in the offset storage (for this post, DynamoDB). This information is used by the driver in the construction of the next micro-batch.

Solution overview

The following diagram shows an example architecture of how to use the connector to read from one Kinesis data stream and write to another.

In this architecture, we use the Amazon Kinesis Data Generator (KDG) to generate sample streaming data (random events per country) to a Kinesis Data Streams source. We start an interactive Spark Structured Streaming session and consume data from the Kinesis data stream, and then write to another Kinesis data stream.

We use Spark Structured Streaming to count events per micro-batch window. These events for each country are being consumed from Kinesis Data Streams. After the count, we can see the results.

Prerequisites

To get started, follow the instructions in the GitHub repo. You need the following prerequisites:

• The AWS Cloud Development Kit (AWS CDK) CLI installed. For instructions, see Install the AWS CDK CLI.
• NPM installed.
• AWS permissions to run AWS CDK commands and deploy an AWS CloudFormation template.

After you deploy the solution using the AWS CDK, you will have the following resources:

• An EMR cluster with the Kinesis Spark connector installed
• A Kinesis Data Streams source
• A Kinesis Data Streams sink

Create your Spark Structured Streaming application

After the deployment is complete, you can access the EMR primary node to start a Spark application and write your Spark Structured Streaming logic.

As we mentioned earlier, you use the new open source Kinesis Spark connector to consume data from Amazon EMR. You can find the connector code on the GitHub repo along with examples on how to build and set up the connector in Spark.

In this post, we use Amazon EMR 7.1, where the connector is natively available. If you’re not using Amazon EMR 7.1 and above, you can use the connector by running the following code:

cd /usr/lib/spark/jars 
sudo wget https://awslabs-code-us-east-1.s3.amazonaws.com/spark-sql-kinesis-connector/spark-streaming-sql-kinesis-connector_2.12-1.2.1.jar
sudo chmod 755 spark-streaming-sql-kinesis-connector_2.12-1.2.1.jar

Complete the following steps:

1. On the Amazon EMR console, navigate to the emr-spark-kinesis cluster.
2. On the Instances tab, select the primary instance and choose the Amazon Elastic Compute Cloud (Amazon EC2) instance ID.
   

You’re redirected to the Amazon EC2 console.

3. On the Amazon EC2 console, select the primary instance and choose Connect.
   
4. Use Session Manager, a capability of AWS Systems Manager, to connect to the instance.
   
5. Because the user that is used to connect is the ssm-user, we need to switch to the Hadoop user:

   sudo su hadoop

   

6. Start a Spark shell either using Scala or Python to interactively build a Spark Structured Streaming application to consume data from a Kinesis data stream.

For this post, we use Python for writing to a stream using a PySpark shell in Amazon EMR.

7. Start the PySpark shell by entering the command pyspark.
   

Because you already have the connector installed in the EMR cluster, you can now create the Kinesis source.

8. Create the Kinesis source with the following code:

   kinesis = spark.readStream.format("aws-kinesis") \
       .option("kinesis.region", "<aws-region>") \
       .option("kinesis.streamName", "kinesis-source") \
       .option("kinesis.consumerType", "GetRecords") \
       .option("kinesis.endpointUrl", "https://kinesis.<aws-region>.amazonaws.com") \
       .option("kinesis.startingposition", "LATEST") \
       .load()

For creating the Kinesis source, the following parameters are required:

• Name of the connector – We use the connector name aws-kinesis
• kinesis.region – The AWS Region of the Kinesis data stream you are consuming
• kinesis.consumerType – Use GetRecords (standard consumer) or SubscribeToShard (enhanced fan-out consumer)
• kinesis.endpointURL – The Regional Kinesis endpoint (for more details, see Service endpoints)
• kinesis.startingposition – Choose LATEST, TRIM_HORIZON, or AT_TIMESTAMP (refer to ShardIteratorType)

For using an enhanced fan-out consumer, additional parameters are needed, such as the consumer name. The additional configuration can be found in the connector’s GitHub repo.

kinesis_efo = spark \
.readStream \
.format("aws-kinesis") \
.option("kinesis.region", "<aws-region>") \
.option("kinesis.streamName", "kinesis-source") \
.option("kinesis.consumerType", "SubscribeToShard") \
.option("kinesis.consumerName", "efo-consumer") \
.option("kinesis.endpointUrl", "https://kinesis.<aws-region>.amazonaws.com") \
.option("kinesis.startingposition", "LATEST") \
.load()

Deploy the Kinesis Data Generator

Complete the following steps to deploy the KDG and start generating data:

1. Choose Launch Stack:
   

You might need to change your Region when deploying. Make sure that the KDG is launched in the same Region as where you deployed the solution.

2. For the parameters Username and Password, enter the values of your choice. Note these values to use later when you log in to the KDG.
3. When the template has finished deploying, go to the Outputs tab of the stack and locate the KDG URL.
4. Log in to the KDG, using the credentials you set when launching the CloudFormation template.
5. Specify your Region and data stream name, and use the following template to generate test data:

   {
       "id": {{random.number(100)}},
       "data": "{{random.arrayElement(
           ["Spain","Portugal","Finland","France"]
       )}}",
       "date": "{{date.now("YYYY-MM-DD hh:mm:ss")}}"
   }

   

6. Return to Systems Manager to continue working with the Spark application.
7. To be able to apply transformations based on the fields of the events, you first need to define the schema for the events:

   from pyspark.sql.types import *
   
   pythonSchema = StructType() \
    .add("id", LongType()) \
    .add("data", StringType()) \
    .add("date", TimestampType())

8. Run the following the command to consume data from Kinesis Data Streams:

   from pyspark.sql.functions import *
   
   events= kinesis \
     .selectExpr("cast (data as STRING) jsonData") \
     .select(from_json("jsonData", pythonSchema).alias("events")) \
     .select("events.*")

9. Use the following code for the Kinesis Spark connector sink:

   events \
       .selectExpr("CAST(id AS STRING) as partitionKey","data","date") \
       .writeStream \
       .format("aws-kinesis") \
       .option("kinesis.region", "<aws-region>") \
       .outputMode("append") \
       .option("kinesis.streamName", "kinesis-sink") \
       .option("kinesis.endpointUrl", "https://kinesis.<aws-region>.amazonaws.com") \
       .option("checkpointLocation", "/kinesisCheckpoint") \
       .start() \
       .awaitTermination()

You can view the data in the Kinesis Data Streams console.

10. On the Kinesis Data Streams console, navigate to kinesis-sink.
11. On the Data viewer tab, choose a shard and a starting position (for this post, we use Latest) and choose Get records.

You can see the data sent, as shown in the following screenshot. Kinesis Data Streams uses base64 encoding by default, so you might see text with unreadable characters.

Clean up

Delete the following CloudFormation stacks created during this deployment to delete all the provisioned resources:

• EmrSparkKinesisStack
• Kinesis-Data-Generator-Cognito-User-SparkEFO-Blog

If you created any additional resources during this deployment, delete them manually.

Conclusion

In this post, we discussed the open source Kinesis Data Streams connector for Spark Structured Streaming. It supports the newer Data Sources API V2 and Spark Structured Streaming for building streaming applications. The connector also enables high-throughput consumption from Kinesis Data Streams with enhanced fan-out by providing dedicated throughput up to 2 Mbps per shard per consumer. With this connector, you can now effortlessly build high-throughput streaming applications with Spark Structured Streaming.

The Kinesis Spark connector is open source under the Apache 2.0 license on GitHub. To get started, visit the GitHub repo.

About the Authors

Francisco Morillo

Umesh Chaudhari