In this video, you’ll see how to process Amazon Kinesis data streams using AWS Lambda.
With this solution, you can build complex processing logic more easily,
pay only for what you use with a 100 percent serverless model,
and get infinite scalability with no operational overhead.
For this demonstration, we’ll build a simple application that has a source
writing data to a Kinesis data stream.
Then, we’ll use a Lambda function to read data from the Kinesis data stream
and write it into an Amazon DynamoDB table.
A Kinesis data stream is integrated with Lambda through a resource
called an event source mapping.
Typically, when data stored in a Kinesis data stream is consumed by an application,
it is that consumer's responsibility to track-or checkpoint- a stream's last
known position that was successfully processed.
The event source mapping takes care of checkpointing for you, so your Lambda
function code can purely focus on the business logic for processing an event or set of events.
The event source mapping also has a retry mechanism built in, so if your event processing
code fails, it is executed again with the same set of records until it is successful.
If poison messages get into a data stream, they can break the processing
code in a Lambda function.
You can configure the event source mapping to write any poison messages
in a stream to an Amazon Simple Queue Service (Amazon SQS) queue or to an Amazon
Simple Notification Service (Amazon SNS) topic and continue processing the
remaining messages in the stream.
The event source mapping has functionality to control how many
concurrent Lambda functions are used to process a data stream.
It enables you to run up to 10 concurrent Lambda functions per partition
and retains the order of delivery for messages with the same partition key.
The event source mapping can also batch incoming data before it's
delivered to the Lambda function.
You can designate batches by size or window.
You can also specify a tumbling window to accumulate records for a fixed period
and dispatch all those records into a Lambda function at once.
These benefits help cut costs and simplify the process of building
sophisticated event streaming logic, so you can focus on developing the
business logic for processing events in your Kinesis data stream.
Now let’s go to the AWS Management Console and build this application.
We’ll begin by navigating to the Kinesis console, where we’ll create a data stream.
Let’s name our data stream events.
We can choose either On-demand or Provisioned mode for our data stream capacity.
We are going to work with a very small workload with predictable throughput,
so we'll use Provisioned mode.
Let’s keep the number of provisioned shards set to 1 and create the data stream.
Now that our data stream is ready, let’s navigate to DynamoDB and create a table.
We’ll give the table a name and provide a partition key.
Let’s retain the default table settings and create the table.
The DynamoDB table has been created.
Now let’s navigate to the Lambda console and create a function.
We’ll use the option to author from scratch and give the function a name.
Let’s use Python 3.9 to write our function.
We’ll use the default architecture.
Next, let’s configure our Lambda function with the necessary permissions.
We'll use an existing IAM role.
This IAM role has permissions to read from Kinesis and write to DynamoDB.
Now let’s create the function.
Once the function has been created, we can configure its event processing logic.
Let’s start by removing the ToDo Implement placeholder.
Next, we’ll import boto3 so that we can use the AWS SDK for Python.
Next, we’ll create a DynamoDB resource instance.
Next, we’ll call a function named table in the DynamoDB resource and specify
our table name, which is events.
Since the records will be dispatched in batches when our Lambda function is
integrated to Kinesis, our Lambda function handler could receive one or more events.
Let’s write code to handle all the records dispatched to the function handler.
We’ll create a record and extract the partition key from it.
Next, we’ll extract the message payload from the record in Kinesis.
When records are written into Kinesis, the message payload is base64 encoded.
In order to read this encoded stream as a piece of text, we first need to
decode it using a basic default code.
Let’s import base64.
Let’s construct a variable called msg by reading the decoding property in our
record object and passing a record.
This returns a message variable with all the bytes decoded from the base64 stream.
Now let’s write the record to the DynamoDB table.
The table will include a PK column with the partition key for the record.
It will also include a data column with the data stream decoded from bytes into a UTF-8 string.
Let’s configure the function to return status code 200 once all the records have been processed.
Now let’s deploy our Lambda function.
For the function to process items from our Kinesis data stream, we need to
create an event source mapping.
To do this, let’s add a trigger.
We’ll set the source to Kinesis and select our stream named events.
We can specify an EFO consumer for better throughput.
In this case, we’ll skip this option.
Let’s keep the Activate trigger check box selected so that the trigger is activated
as soon as we finish creating it.
Next, we’ll reduce the batch size to 10 because we won’t be writing data at a very fast pace.
We’ll keep the starting position in the stream to latest.
We’ll also leave the default batch window so that the event source mapping will dispatch results as fast as records are received by our Lambda function.
Now let’s add the trigger.
The trigger now appears in the Lambda function overview, and it has been enabled.
The details show that no records have been processed.
This is because we haven’t written any data into our data stream yet.
Let's do that now.
We’ll use the AWS CLI to put a record into our Kinesis data stream.
The record has been written to shard ID 0, and the sequence number for the record is returned.
Now let’s go back to the Lambda console.
We’ll refresh the list to view the updated details for the trigger.
The last processing result is OK, which means the trigger has processed an item.
To determine if the Lambda function produced the results that we want,
let’s navigate to our DynamoDB table.
The primary key and data now appear in our table as expected.
You’ve just seen how to process Amazon Kinesis data streams using AWS Lambda.
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.