Build serverless Event Driven Architectures (EDA) by combining Apache Kafka® with AWS Lambda functions. Learn how to trigger Lambda functions based on events flowing in an Apache Kafka topic
Despite a number of tools available to perform stream processing like Apache Kafka® Streams or Apache Flink®, sometimes you want to write some serverless code or functions to process events. You can do this with AWS Lambda, letting you concentrate on wiring the parsing logic without having to worry about managing or scaling the compute.
In this tutorial we'll cover the basics of how to invoke a Lambda function using an Apache Kafka trigger.
We'll use AWS Lambda triggers, in particular the Apache Kafka trigger, to consume from a specific Kafka topic and invoke the Lambda function on every event (or batch of events).
Lambda's Apache Kafka trigger works with any Apache Kafka cluster, for simplicity this tutorial will showcase how to create one with Aiven by:
Click Skip this step to jump directly to the Apache Kafka service details while the service builds in the background. Enable the Apache Kafka REST API (Karapace) toggle so we can messages in a Kafka topic using the Aiven Console.
When the cluster is up and running, navigate to the Topics tab and create a topic called test. Click on the topic name to view it, then click Messages to consume and produce messages in the topic.
The next step is to define a secure link between Aiven for Apache Kafka and the AWS Lambda function. Aiven for Apache Kafka offers a client certificate authentication method out of the box we can store the secrets in AWS Secrets Manager by:
Loading code...
The <ACCESS_CERTIFICATE> and <ACCESS_KEY> are respectively the Access certificate and Access key you can find in the Aiven Console, under the Aiven for Apache Kafka service Overview tab.
The access certificate should start with -----BEGIN CERTIFICATE----- and end with -----END CERTIFICATE-----, the whole content should be included.
The access key should start with -----BEGIN PRIVATE KEY----- and end with -----END PRIVATE KEY-----, the whole content should be included.
prod/AppLambdaTest/Kafka), a description, tags and resource permissionsThe secret contains the Aiven for Apache Kafka access certificate and key. We need to do the same set of steps to store the CA certificate. The JSON format for this second secret is:
Loading code...
The <CA_CERTIFICATE> is the Access CA you can find in the Aiven Console, under the Aiven for Apache Kafka service Overview tab. You need to include everything from the -----BEGIN CERTIFICATE----- to the -----END CERTIFICATE----- including the two opening and closing strings.
Name the CA related secret prod/AppLambdaTest/KafkaCA
The next step is to define the Lambda function to perform some simple logging of the events coming in. We can create a Lambda function by:
Python 3.11 as Runtimex86_64 as Architecture
The basic role created during the Lambda definition allows only CloudWatch logging by default. We need to enable the role to read the secrets containing the Apache Kafka certificates needed to connect to the Aiven service. We can do that by:
<ROLENAME> link at the bottom of the page to edit the newly created role (the <ROLENAME> is a string including the Lambda function name, the role string and a random string suffix), this opens AWS Identity and Access Management (IAM)GetSecretValueThe new inline policy is added to the newly created role.
Back on the AWS Lambda service page where the newly created function has now all the privileges to access the secrets needed to connect to Apache Kafka. We can setup the trigger by:
1 as batch size, so the function is called on every eventtest as topic nameCLIENT_CERTIFICATE_TLS_AUTH as AuthenticationThe last step in our journey is to define what the Lambda function should do. To define that we need to:
Loading code...
In the above code we:
json to parse the JSON values to push to the kafka topicpartition=list(event['records'].keys())[0]. To retrieve the name of the partition, check the details of an example event in the AWS documentationdata['value'], then print the pizza and name fields, after decoding from base64.It's time to run the function with the Deploy button.
To check if the pipeline is working, we need to:
test topicWe can test it by:
test topicLoading code...
To check that the Lambda function works:
Loading code...
AWS Lambda functions are a widely adopted by developers to ingest and parse a scalable amount of data without having to pre-provision a dedicated service.
The combination between Lambda functions and Apache Kafka offers an interesting option for building scalable, serverless, event driven architectures. It lets you read data from Apache Kafka without having to code an application or deploy a service specific to that. This reduces your overall infrastructure cost and the complexity of your system.
{
"certificate":"<ACCESS_CERTIFICATE>",
"privateKey":"<ACCESS_KEY>"
} {
"certificate":"<CA_CERTIFICATE>"
}import json
import base64
def lambda_handler(event, context):
partition=list(event['records'].keys())[0]
message=event['records'][partition]
for data in message:
stringvalue = base64.b64decode(data['value'])
value=json.loads(stringvalue)
print('pizza ' + value['pizza'])
print('name ' + value['name']){
"pizza":"Pepperoni",
"name": "Frank"
}pizza Pepperoni
name Frank