aws-samples/amazon-qldb-streams-dmv-sample-lambda-python

A Basic Sample to demonstrate how to use Amazon QLDB Streams

Amazon QLDB Streams Sample Lambda in Python

The sample in this project demonstrates how to consume Amazon QLDB Streams. It consists of an AWS Lambda function and is modeled around the Amazon QLDB DMV Sample App. The sample app sends out a message to an SNS Topic (which has an SQS queue subscribed to it)for new records inserted in Person and VehicleRegistration Table.

*Note: This sample can be run with any of the implementations of QLDB Sample App (Java, Python, Typescript). This sample is agnostic of the programming language in which the QLDB Sample App has been implemented because it doesn't interact directly with the QLDB DMV sample app. It simply consumes data that is streamed from your ledger. *

Requirements

Basic Configuration

This sample assumes that you have already setup the Amazon QLDB SMV Sample App. It is recommened to go through the tutorial regarding the DMV Sample App.

In case you don't have it setup, Please refer QLDB DMV Sample App Prequisistes.

SAM CLI

AWS SAM provides you with a command line tool, the AWS SAM CLI, that makes it easy for you to create and manage serverless applications. You need to install and configure a few things in order to use the AWS SAM CLI. See AWS SAM CLI Installation for details.

AWS CLI

SAM requires an S3 bucket to host the source code for Lambda function. We will be using the AWS CLI for creating the bucket. Please read AWS CLI Configuration for help on how to configure the CLI.

Python 3.4 or above

The examples require Python 3.4 or above. Please see the link below for more detail to install Python:

• Python Installation

Deploying the Lambda Stack.

This step assumes that you already have the Amazon QLDB DMV Sample App running. It is expected that vehicle-registration ledger and Person, VehicleRegistration tables have been created.

Note: In case you don't have that running, please follow the Tutorial or the README and complete the steps to create ledger and tables.

1. We would need to create an S3 bucket. This S3 bucket would be used by SAM to host the source code of the Lambda function.

export BUCKET_NAME=some_unique_valid_bucket_name
aws s3 mb s3://$BUCKET_NAME

2. Build the Lambda source code and generate deployment artifacts that target Lambda's execution environment.

sam build

3. Package the Lambda function to S3

sam package \
    --output-template-file packaged.yaml \
    --s3-bucket $BUCKET_NAME

4. Deploy the Lambda stack

sam deploy \
    --template-file packaged.yaml \
    --stack-name STACK_NAME \
    --capabilities CAPABILITY_NAMED_IAM

Replace STACK_NAME with a stack name of your choice

6. Create QLDB Stream

• Sign in to the AWS Management Console, and open the Amazon QLDB console at https://console.aws.amazon.com/qldb.

• In the navigation pane, choose Streams.

• Choose Create QLDB stream.

• On the Create QLDB stream page, enter the following settings:

  • Ledger – Select the ledger vehicle-registration from the drop down.

  • Start date and time – Leave this as the default. The default is current time.

  • End date and time – This can be left blank

  • Destination stream for journal data – Click browse and select RegistrationNotificationStreamKinesis.

  • Enable record aggregation in Kinesis Data Streams – Enables QLDB to publish multiple stream records in a single Kinesis Data Streams record. To learn more, see KPL Key Concepts.

  • IAM role – Select RegistrationNotifierKinesisRole from the dropdown

• When the settings are as you want them, choose Create QLDB stream.

• If your request submission is successful, the console returns to the main Streams page and lists your QLDB streams with their current status.

Triggering the Stream Consumer (Lambda function)

The Lambda function will get triggered for an insert in Person and VehicleRegistration table and will send and an Email through AWS SNS.

To trigger the lambda, insert some sample data to the Person and VehicleRegistration tables. This can be done easily by running the following in DMV Sample App.

python insert_document.py

Refer Inserting Documents.

After inserting documents into the tables, you should see messages in SQS queue named RegistrationNotificationQueue regarding inserts in Person and VehicleRegistration Table. You can use Queue Actions -> View/Delete Messages to see the messages.

Note

• QLDB streams provide an at-least-once delivery guarantee. Each data record that is produced by a QLDB stream is delivered to Kinesis Data Streams at least once. The same records can appear in a Kinesis data stream multiple times. So you must have deduplication logic in the consumer application layer if your use case requires it. For the sake of simplicity, this sample does not implement deduplication.

• This sample does not focus on best practices around using AWS Lambda with Amazon Kinesis. For the sake of simplicity, this sample has configured ShardCount:1 for the AWS Kinesis Stream. For guidance on fine tuning configurations, check here.

Unit Tests

Tests are defined in the tests folder in this project. Use PIP to install the pytest and run unit tests.

pip install pytest pytest-mock --user
python -m pytest tests/ -v

Cleanup

To delete the sample application that you created, use the AWS CLI. Assuming you used your project name for the stack name, you can run the following:

aws cloudformation delete-stack --stack-name STACK_NAME

License

This library is licensed under the MIT-0 License.

amazon-qldb-streams-dmv-sample-lambda-python's Issues

Error when handling stream events with Lambda

When this Lambda is triggered from a Kinesis stream, it results in the error:

[ERROR] KeyError: 'Records'

This is because the Records key should be records. All keys seem to be lowercase. Here is the event object that is being passed into Lambda (clipped):

{'invocationId': '30a92d49-cb1f-496c-b580-d46b3dd38f43', 'sourceKinesisStreamArn': 'arn:aws:kinesis:us-east-1:643292708081:stream/ion-lab-qldb-stream', 'deliveryStreamArn': 'arn:aws:firehose:us-east-1:643292708081:deliverystream/ion-lab-qldb-stream-json', 'region': 'us-east-1', 'records': [{'recordId': '49612811534049402823879335600677040555309725205909209090000000', 'approximateArrivalTimestamp': 1605822934247, 'data': ...}