Building a serverless architecture for data collection with AWS Lambda

AWS Lambda is one of the best solutions for managing a data collection pipeline and for implementing a serverless architecture. In this post, we’ll discover how to build a serverless data pipeline in three simple steps using AWS Lambda Functions, Kinesis Streams, Amazon Simple Queue Services (SQS), and Amazon API Gateway!

How to build a serverless data pipeline in 3 steps

Data generated by web and mobile applications is usually stored either to a file or to a database (often a data warehouse). Because the data comes from different sources, such as the back-end server or the front-end UI, it is often very heterogeneous. Furthermore, this data must typically be synced with multiple services such as a CRM (e.g., Hubspot), an email provider (e.g., Mailchimp), or an analytics tool (e.g., Mixpanel). Moreover, this process must be backward compatible and avoid data loss. For such reasons, building and managing a custom infrastructure that is able to handle the collection of such data could be very expensive.

I would like to propose a solution based on AWS Cloud services with the goal of creating a serverless architecture. This also fits a typical use case where the data flow involves thousands or more events per minute.

The following schema summarizes the architecture that we’ll be describing:

Serverless Architecture for Data Collection Pipeline with AWS Lambda Functions and Kinesis Streams
Serverless Architecture for Data Collection Pipeline with AWS Lambda Functions and Kinesis Streams

The use case that I will be analyzing involves the collection of data from multiple sources: backend, front-end, and mobile generated events. The architecture for data collection is designed to send events to different destinations, like our data warehouse and other third-party services (e.g., Hubspot, Mixpanel, GTM, customer.io). As a result, the architecture includes several serverless AWS cloud services, creating a basic data collection flow that can be easily extended by adding further modules as needed:

  1. Amazon Kinesis Streams collects events that originate from front-end and mobile applications through an Amazon API Gateway that works as a REST endpoint. The events generated by the server are directly sent to Kinesis Streams using the Kinesis Producer Library.
  2. AWS Lambda Functions are used to manage the core logic of the pipeline. A first Lambda Function consumes the output of Kinesis Streams and then forwards events to each single custom Lambda Function, which works as a connector to a third-party service, providing additional features such as authentication and the retry strategy.
  3. Amazon SQS queues are used as a DLQ (Dead Letter Queue) for each AWS Lambda Function as a fallback in case of multiple failures in data processing.

Architecture implementation

1. Amazon Kinesis Streams: Manage events from multiple sources

Let’s analyze the architecture in detail.
The data that we need to track is generated by the back-end server, the front-end website, and the mobile application. The back-end directly sends events to Kinesis Streams using the KPL (Kinesis Producer Library). An example of server-side generated events is a user login or a file download.

On the other hand, events originated from the front-end and from mobile are sent to Kinesis Streams through an Amazon API Gateway that exposes a rest endpoint. An example of such events is a page scroll in the homepage tracked via javascript, or a button clicked in the iOS or Android app.

First of all, why Amazon Kinesis Streams? Amazon Kinesis Streams is the ideal tool if you need to collect and process large streams of data in real time. As a result, there are typical scenarios for using Amazon Kinesis Streams:

  • Managing multiple producers that push their data feeds directly into the same stream
  • Collecting real-time analytics and metrics
  • Processing application logs
  • Integrating the data collection pipeline with other AWS services (i.e. consumers) to process the events

In addition, the API Gateway can be used as the endpoint for front-end and mobile. During the configuration of Amazon Kinesis Streams, we suggest starting with a single shard and increase it only when needed.

The following python code allows you send 300 events to the configured stream. The attribute PartitionKey is a random id used to partition events among shards; in the example, we generated it using a datetime field. Finally, pay attention to payload content format: remember to always add data and name fields. You can find a ready-to-use blueprint in the AWS Lambda console.

import boto3
from time import gmtime, strftime
client = boto3.client(
    service_name="kinesis",
    region_name="us-east-1",
)
for i in xrange(300):
    print "sending event", i, "\nresponse: ",
    response = client.put_record(
        StreamName="data-collection-stream",
        Data='{"name":"event-%d","data":{"square":%d}}' % (i, i*i),
        PartitionKey=strftime("PK-%Y%m%d-%H%M%S", gmtime()),
    )
    print response

 2. Manage events routing and retry strategy with AWS Lambda Functions

AWS Lambda is a serverless cloud service by Amazon Web Services that runs your code for virtually any type of backend application. Lambdas automatically manage the underlying compute resources for you in response to events triggered from other AWS services or call it directly from the outside. Relevant AWS Lambda Functions:

  • Process a single event in real-time without managing servers
  • Are highly scalable
  • Provide a fallback strategy in case of error

With reference to the proposed architecture, an AWS Lambda Function is first directly triggered by Kinesis Streams itself. To preserve the priority, we recommend configuring the trigger with batch size equals to 1 and starting position set to trim the horizon, as shown in the following snapshot of the AWS Lambda Function console:

Trigger AWS Lambda Functions with Amazon Kinesis Streams
Trigger AWS Lambda Functions with Amazon Kinesis Streams

This AWS Lambda Function is in charge of routing the events coming from the Kinesis Stream to several destination services. For a more cost-effective solution, we recommend implementing a conditional routing based on the same event properties. For instance, based on the event name, a set of rules can be configured to decide whether to discard or to forward an event toward a certain destination.
With this purpose, I defined a set of routing rules in a JSON file stored in an S3 bucket:

[
 {
   "destination_name": "mixpanel",
   "destination_arn": "arn:aws:lambda:region:account-id:function:function-name:prod",
   "enabled_events": [
     "page_view",
     "search",
     "button_click",
     "page_scroll",
   ]
 },
 {
   "destination_name": "hubspotcrm",
   "destination_arn": "arn:aws:lambda:region:account-id:function:function-name:prod",
   "enabled_events": [
     "login",
     "logout",
     "registration",
     "page_view",
     "search",
     "email_sent",
     "email_open",
   ]
 },
 {
   "destination_name": "datawarehouse",
   "destination_arn": "arn:aws:lambda:region:account-id:function:function-name:prod",
   "enabled_events": [
     "login",
     "logout",
     "registration",
     "page_view",
     "search",
     "button_click",
     "page_scroll",
     "email_sent",
     "email_open",
   ]
 }
]

I recommend using the AWS Lambda Function environment variables to configure the name of the S3 file and bucket so that they can be updated at any time without editing the code. Each rule has a destination_arn attribute that configures where the event is to be sent if its name is included in the list of enabled_events. The AWS Lambda Function will send the event to the configured ARN (Amazon Resource Name), with each corresponding to a specific Lambda Function.

I suggest outputting events one by one to avoid data loss. In this way, the AWS Lambda Function in charge of conditional routing is sure to manage only one message at a time.

Finally, we need to implement a new AWS Lambda Function that is in charge of validating the incoming event and forwards it to each destination service (e.g., Google Analytics, Mixpanel, Hubspot, database). Such functions are asynchronously invoked by the routing Lambda (using the destination_arn).
Each AWS Lambda function implements:

  • The connector: It provides the logic to connect to the destination service (e.g., HubSpot)
  • The retry strategy: If event saving fails, it retries for a specified number of times with an exponential delay

In addition, a few suggestions that we found useful during the design and the development of the data collection architecture:

  • Invoke Lambda Functions that work as connector asynchronously.
  • Always create aliases and versions for each Function. This configuration enables you to publish a different version of your Lambda Function code depending on development, beta, and production workflow.
  • Use environment variables for configurations.
  • Create a custom IAM role for each AWS Lambda Function.
  • Detect delays in stream processing by monitoring the IteratorAge metric in the Lambda console’s monitoring tab.

 3. Configuring a Dead Letter Queue on AWS Lambda to avoid event loss

While each Lambda Function implements its own retry strategy, some events may not be successfully stored by the destination service (e.g., network problems, missing data, etc.).

Fortunately, Lambda allows you to implement a fallback strategy and, in the event of errors, you can discard the event and store it into a so-called Dead Letter Queue (DLQ) The DLQ can be either an SQS (Amazon Simple Queue Service) or an SNS (Amazon Simple Notification Service).

From advanced settings of Lambda (see the following snapshot) you can configure the AWS Lambda Function to forward payloads that were not processed to a dedicated Dead Letter Queue. In our example, we configured an SQS queue.

Configure DLQ in AWS Lambda to avoid data loss in our Serverless architecture
Configure DLQ in AWS Lambda to avoid data loss in our serverless architecture

You can recover the complete list of DLQ messages using another AWS Lambda Function that is manually triggered. Remember to set a DLQ on each Lambda Function that can fail! Then you can process all of the collected events again with a custom script, like the following:

import json
import boto3
def get_events_from_sqs(
       sqs_queue_name,
       region_name='us-west-2',
       purge_messages=False,
       backup_filename='backup.jsonl',
       visibility_timeout=60):
    """
        Create a json backup file of all events in the SQS queue with the given 'sqs_queue_name'.
        :sqs_queue_name: the name of the AWS SQS queue to be read via boto3
        :region_name: the region name of the AWS SQS queue to be read via boto3
        :purge_messages: True if messages must be deleted after reading, False otherwise
        :backup_filename: the name of the file where to store all SQS messages
        :visibility_timeout: period of time in seconds (unique consumer window)
        :return: the number of processed batch of events
    """
    forwarded = 0
    counter = 0
    sqs = boto3.resource('sqs', region_name=region_name)
    dlq = sqs.get_queue_by_name(QueueName=sqs_queue_name)
    with open(backup_filename, 'a') as filep:
        while True:
            batch_messages = dlq.receive_messages(
                MessageAttributeNames=['All'],
                MaxNumberOfMessages=10,
                WaitTimeSeconds=20,
                VisibilityTimeout=visibility_timeout,
            )
            if not batch_messages:
                break
            for msg in batch_messages:
                try:
                    line = "{}\n".format(json.dumps({
                       'attributes': msg.message_attributes,
                       'body': msg.body,
                    }))
                    print("Line: ", line)
                    filep.write(line)
                    if purge_messages:
                        print('Deleting message from the queue.')
                        msg.delete()
                    forwarded += 1
                except Exception as ex:
                    print("Error in processing message %s: %r", msg, ex)
            counter += 1
            print('Batch %d processed', counter)

Conclusions

I strongly recommend that you consider building a serverless architecture for the management of your data collection pipeline. In this post, we designed a serverless data collection pipeline using several AWS Cloud services such as AWS Lambda Functions and Amazon Kinesis Streams in order to ensure scalability and prevent data loss. The main advantage of this solution is the full control of the process and the cost, since you only pay for the compute time that you consume.

In conclusion, I would suggest a couple of very useful extensions that can be integrated into the serverless architecture:

  • Create a custom CloudWatch dashboard to verify the performance of the processing, monitor the presence of events in the DLQ, and eventually raise alerts in the case of issues.
  • Configure Kinesis Firehose and link it as the output of the Kinesis Streams module to ensure that all data is backed up on S3.
  • Implement a listener with Kinesis Analytics to discover correlations among the incoming events.
Avatar

Written by

David Santucci

Computer science engineer, passionate about machine learning and Lego Mindstorms.

Related Posts

Avatar
Jeremy Cook
— September 17, 2019

Cloud Migration Risks & Benefits

If you’re like most businesses, you already have at least one workload running in the cloud. However, that doesn’t mean that cloud migration is right for everyone. While cloud environments are generally scalable, reliable, and highly available, those won’t be the only considerations dri...

Read more
  • AWS
  • Azure
  • Cloud Migration
Joe Nemer
Joe Nemer
— September 12, 2019

Real-Time Application Monitoring with Amazon Kinesis

Amazon Kinesis is a real-time data streaming service that makes it easy to collect, process, and analyze data so you can get quick insights and react as fast as possible to new information.  With Amazon Kinesis you can ingest real-time data such as application logs, website clickstre...

Read more
  • amazon kinesis
  • AWS
  • Stream Analytics
  • Streaming data
Joe Nemer
Joe Nemer
— September 6, 2019

Google Cloud Functions vs. AWS Lambda: The Fight for Serverless Cloud Domination

Serverless computing: What is it and why is it important? A quick background The general concept of serverless computing was introduced to the market by Amazon Web Services (AWS) around 2014 with the release of AWS Lambda. As we know, cloud computing has made it possible for users to ...

Read more
  • AWS
  • Azure
  • Google Cloud Platform
Joe Nemer
Joe Nemer
— September 3, 2019

Google Vision vs. Amazon Rekognition: A Vendor-Neutral Comparison

Google Cloud Vision and Amazon Rekognition offer a broad spectrum of solutions, some of which are comparable in terms of functional details, quality, performance, and costs. This post is a fact-based comparative analysis on Google Vision vs. Amazon Rekognition and will focus on the tech...

Read more
  • Amazon Rekognition
  • AWS
  • Google Cloud Platform
  • Google Vision
Alisha Reyes
Alisha Reyes
— August 30, 2019

New on Cloud Academy: CISSP, AWS, Azure, & DevOps Labs, Python for Beginners, and more…

As Hurricane Dorian intensifies, it looks like Floridians across the entire state might have to hunker down for another big one. If you've gone through a hurricane, you know that preparing for one is no joke. You'll need a survival kit with plenty of water, flashlights, batteries, and n...

Read more
  • AWS
  • Azure
  • Google Cloud Platform
  • New content
  • Product Feature
  • Python programming
Joe Nemer
Joe Nemer
— August 27, 2019

Amazon Route 53: Why You Should Consider DNS Migration

What Amazon Route 53 brings to the DNS table Amazon Route 53 is a highly available and scalable Domain Name System (DNS) service offered by AWS. It is named by the TCP or UDP port 53, which is where DNS server requests are addressed. Like any DNS service, Route 53 handles domain regist...

Read more
  • Amazon
  • AWS
  • Cloud Migration
  • DNS
  • Route 53
Alisha Reyes
Alisha Reyes
— August 22, 2019

How to Unlock Complimentary Access to Cloud Academy

Are you looking to get trained or certified on AWS, Azure, Google Cloud Platform, DevOps, Cloud Security, Python, Java, or another technical skill? Then you'll want to mark your calendars for August 23, 2019. Starting Friday at 12:00 a.m. PDT (3:00 a.m. EDT), Cloud Academy is offering c...

Read more
  • AWS
  • Azure
  • cloud academy content
  • complimentary access
  • GCP
  • on the house
Avatar
Michael Sheehy
— August 19, 2019

What Exactly Is a Cloud Architect and How Do You Become One?

One of the buzzwords surrounding the cloud that I'm sure you've heard is "Cloud Architect." In this article, I will outline my understanding of what a cloud architect does and I'll analyze the skills and certifications necessary to become one. I will also list some of the types of jobs ...

Read more
  • AWS
  • Cloud Computing
Avatar
Nitheesh Poojary
— August 19, 2019

Boto: Using Python to Automate AWS Services

Boto allows you to write scripts to automate things like starting AWS EC2 instances Boto is a Python package that provides programmatic connectivity to Amazon Web Services (AWS). AWS offers a range of services for dynamically scaling servers including the core compute service, Elastic...

Read more
  • Automated AWS Services
  • AWS
  • Boto
  • Python
Avatar
Andrew Larkin
— August 13, 2019

Content Roadmap: AZ-500, ITIL 4, MS-100, Google Cloud Associate Engineer, and More

Last month, Cloud Academy joined forces with QA, the UK’s largest B2B skills provider, and it put us in an excellent position to solve a massive skills gap problem. As a result of this collaboration, you will see our training library grow with additions from QA’s massive catalog of 500+...

Read more
  • AWS
  • Azure
  • content roadmap
  • Google Cloud Platform
Avatar
Adam Hawkins
— August 9, 2019

DevSecOps: How to Secure DevOps Environments

Security has been a friction point when discussing DevOps. This stems from the assumption that DevOps teams move too fast to handle security concerns. This makes sense if Information Security (InfoSec) is separate from the DevOps value stream, or if development velocity exceeds the band...

Read more
  • AWS
  • cloud security
  • DevOps
  • DevSecOps
  • Security
Avatar
Stefano Giacone
— August 8, 2019

Test Your Cloud Knowledge on AWS, Azure, or Google Cloud Platform

Cloud skills are in demand | In today's digital era, employers are constantly seeking skilled professionals with working knowledge of AWS, Azure, and Google Cloud Platform. According to the 2019 Trends in Cloud Transformation report by 451 Research: Business and IT transformations re...

Read more
  • AWS
  • Cloud skills
  • Google Cloud
  • Microsoft Azure