In this video, we are going to do hands on up process kinesis streams using a lambda.
So kinesis streams are very similar with dynamic DB streams when it comes to a process from a lambda
function, you can see the image on the slide.
So basically there is client application.
These are the client application, mobile client, traditional server users, Iot device and so on.
All these client application produced the data and these data comes to Amazon Kinesis data streams and
Amazon Kinesis provides the streams and Adobe is lambda received streams with pole based event source
mapping in location according to batch size and run the function itself.
So there are several steps that we can follow when we provide this architecture.
For example, first of all, we will create a kinesis data streams.
After that, we will create a lambda function, and after that we will add required permissions for
pulling kinase charts from the lambda function.
And after that we will create event source mapping, pulling in location type between Lambda and Kinesis.
And after that we will develop our lambda function code with incoming Kinesis records.
And lastly, we will perform end to end test.
So let's get started.
Remember that everyone hands on loop in the serverless project we have two steps.
The first step is creating infrastructure on a cloud environment.
So that means we will create these hands on lab architecture in a cloud environment.
And the second step is developing lambda function code with interacting kinesis data streams.
So now let me start with the first phase, which is the create infrastructure.
So that means we will create kinesis data streams and Lambda functions for that purpose.
Please open the development console, go to Kinesis, see that Kinesis is collect process and analyze
data stream in the real time.
So we will use Kinesis data stream collecting streaming data and click the create data stream button.
So now I'm going to give the stream name, for example, test stream.
And if you scroll down, you can see the cup estimate we select on demand and you can see the capabilities
of these kinesis data stream and the right capacity and capacity in here.
If you scroll down, you can see the data stream settings.
All these settings are coming from the default, which is the on demand we use in the capacity mode
and leave as it is for all configurations and click to create data stream.
So as you can see that we have created Amazon Kinesis data stream and if you scroll down, you can see
the details in here.
And so to becomes active, if you scroll down, you can see the applications and these applications
split by two.
That means there is a producer applications and also consumer applications.
Since we didn't integrate with any kind of data stream, now there is no producer and consumer, but
we can create producers and consumers using the SDK and other tools you can see in here.
Okay.
If you scroll up and see that there is also enhanced fanart.
If you scroll down, you can see that we here specify consumers using enhanced fanart.
Maybe you remember in the last section we saw an article that explained how we can perform find out
architecture with using Amazon Kinesis data streams and lambda function.
After we have successfully define test stream of Amazon Dynamo data streams, now I'm going to create
a lambda function to perform the create infrastructure phase.
Go to lambda function and create function please and give the name of the function as a kinesis function.
And scroll down.
So this time we are going to pull the records from the kinesis before creating with default.
Let me check, is there any predefined policy in here if I search for the kinesis?
There is no any predefined a list policy for now, so please leave as it is and create any role with
basic learner permission and give the name as a stage function and click the create function button.
So this will take some time and after we have created the lambda function we will check several things,
as you remember, that we will follow the steps into the architecture.
We have finished the first two item and continue with the other ones.
So let's go back to our management console and see what happened.
Okay.
We have successfully define kinase function and now I'm going to follow the other steps, which is the
third step, add required permissions to polling kinase charts from the lambda.
So for that purpose, please go to lambda function, go to configuration and go to permissions.
See that we don't have any kinase permission.
I would like to add kinase related permissions in order to receive items from the kinases, please go
to execution role page and go to permissions.
And if you scroll down, you can see the permission policies.
Basically, we will add a policy which required Kinesis rules if you open the or we use to do code.
You can see that these kinesis items and permissions should be required in order to receive item from
the kinesis, from the lambda function and all these permissions, including a Lambda Kinesis execution
rule predefined policy from the ADA.
Please copy this rule and come here, paste in here and you can see that if you click the add permission
and attach the policy, we can search for the attached policy and there is a management permission and
policy which name is a lambda kinase exceptional?
And if you expand this permission, you can see that all the required stream, required kinesis stream
required permissions actions included in this policy.
And we can select this policy and attach the policy into our item role, which is our execution of lambda
function.
If you scroll down, you can see in here and also if you refresh the page of the lambda, we will verify
that the permissions come here and you can see the kinesis and see that all these actions are allowed
into our lambda function.
So that means our lambda function now capable to receive items from the Amazon Kinesis.
But how we can receive items from the Kinesis with pull based invocations.
For that purpose, we will follow the fourth step.
If you come back to our architecture, you can see that we should create event source mapping, pulling
in location type between London Kinesis.
In order to do that, we are going to add a trigger to our Kinesis function.
If you come here and search for the kinesis, you can see that we can select event source as a kinesis
of lambda function and give the kinesis stream name, which is a test stream.
And then if there is a consumer, we can set the consumer.
We don't have consumer and I'm going to scroll down, see that these are the pull based configuration
that we don't change anything.
I'm leaving as it is and click the red button.
So we are successfully create event source mapping communication between Lambda Function and Kinesis.
So if you come back to our architecture, you can see that the last thing we should do that develop
our lambda function code with incoming kinesis record.
So if you go back to our main console, how we can understand kinesis event JSON to trigger lambda function
the easiest way, go to test window of the lambda function and create a new event which is a test kinesis.
And if you scroll down, you can select the template.
If you search for the Kinesis and search for the Kinesis data streams, if you search for the Kinesis
data streams, you can see that the last item is Kinesis data stream and see the example event JSON
object that will trigger the lambda function.
So if you copy all these codes and go to Visual Studio code, I'm going to create one JSON file and
you can copy paste in here.
Okay.
Very good.
So this is the example event object theory.
Go to lambda function again.
There is a records array and every record item has kinesis and these are the attributes and key properties
that we can define about the kinesis.
You can see the event source is a Kinesis event ID event name and event source and these are the Kinesis
detailed attributes that we can see in the UN JSON object.
Okay, very good.
So if you come back to our lambda function now, we can develop our lambda function according to this
test event.
So if you go back to our pseudocode, I'm going to develop our lambda function in the index case.
See that I have created separate lecture and section folder and go to index and see the implementation
I have already implement before because this is very similar with the dynamic DB systems.
First of all, we log incoming event and after that we will iterate the records and get the record into
the our kinesis stream.
But this time the only difference is this record kinase is stored as a base 64 and we need to decode
this record because it comes with the encoded way.
So for example, if you go to record Kinesis data in the incoming event, JSON example called event
is an example go to record Kinesis data, you can see that these value is encoded base 64 value and
it is required to decode when we consume this data.
So that's why if we when we develop this lambda function, we are basically perform the decode operation
according to base 64 and get the payload.
So after that, I'm going to log the payload.
This is very easy business logic, but you can increase or you can change the business logic.
Applying the final pattern as we talked before.
Okay, very good.
Please develop these indexes and copy these codes and go back to our estimation console and go to Lambda,
select the code and remove the indexes and paste in here.
Click the deploy button.
Look, we have successfully deployed our lambda function and now we are ready for the received data
from the Kinesis.
If you open the architecture, as you can see that we have created the Kinesis data streams and create
lambda function and develop our lambda function and create event source mapping connection between kinase
and lambda function.
The only difference with the dynamic DC stream is that Kinesis using Base64 when transferring data into
it.
So let's test the whole process in the next video.