Subscribe from a Kinesis Data Stream using AWS Lambda
HTML-код
- Опубликовано: 5 авг 2024
- In this video we will learn what Amazon Kinesis is, as well as how to create a Kinesis consumer using an AWS Lambda function.
Timestamps:
- Intro: 0:00
- What is Kinesis?: 0:55
- Create the Kinesis data stream: 5:36
- Create a Lambda Function: 6:44
- Time for testing: 9:54
Code example: github.com/endre-synnes/pytho...
Inspiration for today’s video:
- Medium article: / fine-tuning-your-lambdas
- AWS documentation: docs.aws.amazon.com/streams/l...
Subscribe button animation: touchtechnologyreview.com/sub...
Follow me on Github: github.com/endre-synnes
Hi thank you. This is very useful. I have serached for this kind of explanatory video. Great work. If you can please do a video about limitations related with Kinesis producers and consumers. Like write per second, read per second, time between records, etc...
Thank you! I'm glad you found it helpful :)
Yes, that's something I can look into. Thank you for the tip!
good explain
Thank you! 😄
Hi, can you configure the lambda to only be triggered on specific Kinesis data record types (defined by you).
Hi 😊 Yes, there is several ways to accomplish this. You can for example enable event filtering (docs.aws.amazon.com/lambda/latest/dg/invocation-eventfiltering.html#filtering-syntax) when you create the Lambda Trigger.
Thanks for the great video. I'm new to Pub/Sub and recently I received a project in which the customer wants to integrate 2 APIs. One API will send data and the other needs to consume it to update their systems. I think pub/sub is the right approach. Any suggestions? Am I on the right track?
Thank you!l 😄
Yes, for me it sounds like a pub/sub solution would be a good idea. Then again I don’t know the details of your applications.
There are many ways to implement pub/sub, depending on your requirements. The first thing I would consider, is if asynchronous communication is possible for your APIs to function as intended. By that I mean if the sender needs to know or wait until a message/request has been processed or accepted by the consumer API. If not, then implementing pub/sub becomes a lot easier. Next I would suggest looking into the different services like SQS, SNS, Kinesis to see if they can meet your requirements. For example SNS only broadcasts notifications to every listener once, but are really fast and salable. SQS could be sufficient if you only need a normal queue-service, and Kinesis could be an alternative if you need to do some more advanced stuff like processing the messages before hitting the target destination. All these services do come with some pros and cons. Some have lower latency or can ensure “only-once-delivery”, some are really cheap and others are more expensive. And maybe the most important, some does come with more complexity than others, f.ex. Kafka.
I hope this was somewhat helpful 😄
@@EndreSynnes this was infinitely helpful, thank you so very much! From my understanding the sender API will only send a full load of data the first time around. Subsequent requests to the sender API will provide only delta of the last call - the API is asynchronous. The receiver API needs to update their records accordingly. My only concern is if for whatever reason the receiver processes an old request before a newer one or if the messages get hung and the receiver needs to process multiple messages, but some of the older records are no longer needed to be pushed because a newer version for that record arrived. Lastly, how does the receiver handle messages. For example, it needs to check which records in the data are for update and which ones are for insert. Can't decide between AWS SNS or AWS Kinesis.
I'm glad you found it helpful! :)
Yes there are many things to consider when creating a pub/sub solution. Also, when you say "full load of data'', I'm not sure if this means a big object or not, but keep in mind the size of the messages/records, as the different services have different limits for that as well.
If the order of requests are important, then I guess an SQS fifo queue or Kinesis would be the way to go. A benefit of using SQS and Kinesis is that messages can be stored there (for a limited time) until the consumer is able to read them. Then again requiring the order of messages to be correct will potentially limit the scaling when/if you get more traffic.
If you on the other hand don't need to consider the order of messages and and don't care if you miss some requests (due to the consumer application not being able to consume the requests at the moment of them being sent). Then SNS could be a good solution, as it is really fast and scalable.
As for differing between updates and creation of new objects. I guess you can always treat incoming requests as create requests to a database. Depending on your database and implementation, you could maybe just override values for objects with the same id. Of course if the request is only an update and the update doesn't include all the required fields to create a new record, then that could be an issue if the object is not already in the database. To address this issue you can f.ex. use a dead letter queue ( docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/sqs-dead-letter-queues.html ) and implement some logic to process these (possible) edge cases. Or simply split create/update requests in different queues/streams.
There are many other ways to implement this kind of behavior, so these examples are just what I thought of at the top of my head. Whatever solution you end up choosing, just make sure you don’t over engineer it the first time around, until you know for sure what you need and don’t need. :)
Lambda function to take kinesis data stream json data and put new line delimited and output me to se in parquet, can you help here
And Amazon connect is incoming data to data stream here
Hi, thank you for your question! 😄
I'm not quite sure what you mean. Do you want to see how to process a message coming in from a Kinesis data stream and sending it onwards to another Kinesis data stream?
I mean I want a lambda function code which take json from data stream and put new line delimited json in se as output and lambda to be attached to delivery stream
I am getting this error
{
"errorMessage": "'Records'",
"errorType": "KeyError",
Hi 😄
Without seeing your code, I would assume that this is python raising an exception because It can't find a property matching the key you have provided to extract the kinesis payload.
If you can provide your code I can have a quick look to see if that is the case. 😄