Creating Dataflow Pipelines with Amazon Kinesis

Creating Dataflow Pipelines with Amazon Kinesis

Amazon Kinesis LogoAlthough the recent public preview of Amazon Managed Streaming for Kafka (MSK) certainly made headlines, Kinesis remains Amazon's supported, production, real-time streaming service. In this blog post, I'll show you how to get started using StreamSets Data Collector to build dataflow pipelines to send data to and receive data from Amazon Kinesis Data Streams.

Getting Started with Amazon Kinesis Data Streams

A Kinesis stream is partitioned into shards in much the same way as a Kafka topic is split into partitions. Shards provide scalability, but message order is only preserved within a shard, and not between shards. It's important to remember this if the order in which messages arrive at consumer applications is important!

We can use the Kinesis CLI to create a stream with a specified number of shards. We'll start with a single shard to keep things simple:

Note that the create-stream command returns immediately, but it takes a few seconds for the stream to be created. We use the describe-stream command to monitor progress:

When we use the CLI to add data to the stream, we need to specify a partition key. Records with the same partition key will be sent to the same partition. In this way, we can preserve ordering of messages relating to the same object. For example, we might use ‘account number' as a partition key so that messages concerning the same account are ordered correctly. In our simple example, we only have a single shard, so the partition key is irrelevant for now.

Retrieving messages from Kinesis using the CLI is a little more involved. First, we need to get a shard iterator, specifying an iterator type:

  • TRIM_HORIZON to start iterating from the oldest data record in the shard
  • LATEST to start iterating from just after the most recent record in the shard
  • AT_SEQUENCE_NUMBER to start iterating from a specific record
  • AFTER_SEQUENCE_NUMBER to start iterating from just after a specific record
  • AT_TIMESTAMP to start iterating from a given point in time

Let's get a shard iterator with TRIM_HORIZON, so we can get all the data in the shard:

Now we can use get-records with the shard iterator to actually retrieve some records:

Notice that the data is encoded. You can send any sort of data to Kinesis, text or binary, so the CLI base64-encodes the data so that it is shown in a legible format. We can base64-decode the data using openssl:

The Kinesis CLI is a thin wrapper over the underlying API, and there is no single command to just stream messages from a shard. We can use a bit of bash-scripting and the excellent jq JSON processor to build what we need:

If we send messages from a second terminal window, we'll see them arriving in the first:

Now we can send and receive messages from the command line, let's build our first dataflow pipeline with Data Collector and Kinesis!

Sending Data to Kinesis with StreamSets Data Collector

The basic Data Collector tutorial provides a great starting point to generate some realistic messages. I simply replaced the Local FS destination in the tutorial with the Kinesis Producer destination:

Kinesis Producer Pipeline

I configured the destination to send data to taxi-stream, and to use the credit card type (Mastercard, Visa, etc) as the partitioning key. This allows the pipeline to scale across multiple shards, but keeps transactions for each credit card network in the original order:

Kinesis Destination Configuration

Outgoing records are formatted as JSON:

Kinesis Destination Data Format

We'll create taxi-stream with three partitions, so we can see the effect of the partitioning strategy:

Once the stream is active, we'll write a few records. I used Data Preview, with a ‘Preview Batch Size' of 10 and ‘Write to Destinations and Executors' enabled to send 10 records to Kinesis:

Preview Dialog

We can see the 10 records in the preview user interface:

Preview Kinesis Producer Pipeline

Now we can read the records from Kinesis using our CLI script to verify that the data was correctly sent. I'll abbreviate the records for clarity:

Wait – there are only 8 messages there, but we sent 10! Where are the other two? Remember, we're only pulling records from a single shard, but this stream has three shards. Let's try the second shard:

We found the missing two records. As we would expect, the third shard is empty:

You'll notice that the records in the first shard all have credit card type n/a or Mastercard, while the second shard holds records with credit card type Visa. Records are being sent to partitions based on credit card type, as we configured, so that ordering is preserved for transactions related to a given credit card network.

Now that we're successfully sending records to Kinesis, let's create a consumer pipeline.

Receiving Data from Kinesis with StreamSets Data Collector

I'm going to create a dataflow pipeline to run on Amazon EC2, reading records from the Kinesis stream and writing them to MySQL on Amazon RDS. When Data Collector is running on EC2, you can configure the EC2 instance with an AWS IAM role allowing it to access Kinesis and other resources. In this case, I created an AWS IAM role with a policy allowing access to Kinesis as a consumer. There are several permissions that must be granted for a Kinesis consumer – they are listed in the Kinesis documentation. Note that you must grant permissions in DynamoDB as well as Kinesis.

Here's a simple pipeline to run on EC2, reading messages with the Kinesis Consumer origin and writing them to MySQL with the JDBC Producer destination:

Kinesis Consumer Pipeline

Since the pipeline will be running on EC2, we can leave the AWS credentials blank. Data Collector will inherit the permissions configured in the IAM role associated with the EC2 instance on which it is running:

Kinesis Consumer Configuration

Note that we can configure the initial position at which the pipeline should start. Here we can choose between LATEST, TRIM_HORIZON and AT_TIMESTAMP. The Kinesis Consumer is a multithreaded origin, so we can configure the maximum number of threads that Data Collector will create to ingest data. Data Collector will create multiple threads, up to this limit, to process records concurrently, maximizing record throughput.

The Kinesis Consumer data format matches the destination data format in the producer pipeline:

Kinesis Consumer Data Format

I created separate tables for credit card and cash transactions in MySQL, so I could show how records in the same batch can be written to different tables. The only difference between them is that the credit card table has card type and number:

On examining the sample data, we can see that some of the records have payment types other than CRD (credit card) or CSH (cash), so we configure a precondition on the JDBC Producer destination to ensure that only records with CRD and CSH payment types are send to the database:

JDBC Producer General

The JDBC configuration is straightforward. Note the table name expression – this is evaluated for each record to send it to the correct table. I didn't need to define any field to column mappings, since I deliberately created the MySQL tables with column names matching the record field names.

JDBC Producer Config

Running the pipeline, it consumes the ten records we wrote to Kinesis earlier:

Kinesis Consumer Pipeline Running

We can examine the tables in MySQL to check that the data has been written as expected:

Success!

Now let's stop the consumer pipeline, reset the database by truncating the tables, then run both pipelines to process all of the input data and send it to MySQL via Kinesis:

Kinesis Producer & Consumer Pipelines

We can see that all of the records sent by the producer pipeline were received by the consumer pipeline. Nine records were not inserted into the database; this is test data and some records are deliberately corrupted to show error handling.

Now we have the data in MySQL, we can begin to perform some simple analyses – how does the transaction count and average total transaction amount differ across card and cash transactions?

There are more cash than card transactions, but the card transactions have a greater average value.

Conclusion

Amazon Kinesis is a stable, production platform for real-time messaging, transporting messages via streams and scaling via shards. The Kinesis command line tools provide a useful basis for testing stream processing, while StreamSets Data Collector allows you to easily build dataflow pipelines that send and receive data via Kinesis. The Kinesis Producer destination allows you to control how data is sent to Kinesis shards, and the multi-threaded Kinesis Consumer origin can scale to handle high levels of concurrent throughput.

Download StreamSets Data Collector today and get started sending data via Amazon Kinesis – Data Collector is open source and available in a number of binary formats.

Related Resources

Check out StreamSets white papers, videos, webinars, report and more.

Visit the Resource Library

Related Blog Posts

Receive Updates

Receive Updates

Join our mailing list to receive the latest news from StreamSets.

You have Successfully Subscribed!

Pin It on Pinterest