Serverless ETL using AWS Lambda, Pandas, PostgreSQL, AWS CDK
Tags: serverless, aws, cdk, python
Published: 2022-06-25 16:55:12



![cloud infrastructure](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/5ml5jw3y1pm98q8g97dl.png)


## Table of contents
  * [Introduction](#Introduction)
  * [Compute](#Compute)
  * [Infrastructure as code](#Infrastructure as Code)
  * [Continuous delivery](#Continuous Delivery)
  * [Conclusion](#Conclusion)


## Introduction
Created in response to the #ACloudGuruChallenge posted by Forrest Brazeal.

Modern data infrastructure has become so exciting and with limitless possibilities, Today we explore a use case where we pull, transform and visualize covid-19 data from several sources in a serverless and extremely quick to deploy fashion using python as our tool of choice



### Outcome
Having a job that downloads data about covid-19 for case counts, deaths and recoveries for every given day, and stores them in a database for later access and reporting


### Tech stack
**Compute**: Python application on AWS Lambda
**Storage**: S3
**Database**: PostgreSQL on RDS
**Reporting**: Amazon Quicksight
**IaaC and CI/CD**: Github actions + AWS CDK



### Data Sources
We pull data from [This link](https://github.com/nytimes/covid-19-data/blob/master/us.csv) and this [John Hopkins dataset](https://raw.githubusercontent.com/datasets/covid-19/master/data/time-series-19-covid-combined.csv)




## Compute

AWS Lambda is a serverless platform for running small snippets of code in the cloud, in this section we will go over the code used to achieve the end goal

```python
import boto3  
import requests  
import pandas as pd  # Creating a dataframe to store our data
import io  
import json  
import os  # accessing environment variables 
from db import get_latest_date, instantiate_db, load_to_db  # custom modules, will be discussed below
from data_transformation import join_dfs, change_to_datetime, clean_dataframe, rename_cols 

```

And now we discuss the functions defined in our handler


```python

# Instantiate our clients
s3_client = boto3.client('s3')
s3_resource = boto3.resource('s3')
sns_client = boto3.client('sns')

def download_file(url, file_name):
    ''' Downloads data from a file and saves it to s3 as an object'''
    body = requests.get(url).content.strip(b'\n')
    bucket_name = os.environ['S3_BUCKET']
    s3_path = "data/" + file_name
    s3_resource.Bucket(bucket_name).put_object(Key=s3_path, Body=body)
    
    
def load_data(key):
    ''' Loads data from s3 into a pandas Dataframe'''
    obj = s3_client.get_object(Bucket=os.environ['S3_BUCKET'], Key=key)
    csv_file = obj['Body'].read().strip(b'\n')
    df = pd.read_csv(io.BytesIO(csv_file), low_memory=False)
    return df
    
def send_db(df):
    ''' Sends the data to the database that only has a date higher than
    current latest in the db'''
    latest = pd.to_datetime(get_latest_date())
    data_to_send = df[df['Date'] > latest]
    data_to_send.apply(lambda x: load_to_db(x), axis=1)
    return len(data_to_send)
    
def post_to_sns(message):
    ''' Sends the output of the job to an SNS topic which fans out
    the result to any number of interested message consumers'''
    response = sns_client.publish(
        TargetArn=os.environ['SNS_TOPIC'],
        Message=json.dumps({'default': json.dumps(message)}),
        MessageStructure='json'
    )
    return response
```

Our workflow starts with instantiating boto3 clients for communicating with other AWS services, its best practice to start your clients outside of the handler to speed up the execution of your function, refer to Lambda best practices [Here](https://docs.aws.amazon.com/lambda/latest/dg/best-practices.html)

Then we define functions to handle downloading the data and saving it to a storage platform, it's also a best practice to not hard code your s3 bucket name into you functions. especially in our case because we use infrastructure as code that creates the bucket on runtime, we don't know the bucket name in advance

We then have a function for loading our final dataframe into the database

```python
def send_db(df):
    ''' Sends the data to the database that only has a date higher than
    current latest in the db'''
    latest = pd.to_datetime(get_latest_date())
    data_to_send = df[df['Date'] > latest]
    data_to_send.apply(lambda x: load_to_db(x), axis=1)
    return `len(data_to_send)`
```

Let's discuss this function in detail, this job runs daily and so we only want to post the new days into the database instead of loading the entire data everytime, to handle this we query the database for its current latest date. and we apply `load_to_db(record)` only if this record has a date higher than the latest. 

this way we also have unified logic for initial loading or daily runs. and in the end we want to know how many records we processed by returning `len(data_to_send)`

Lastly, we define a function that posts the results of our run to SNS. we will see how we use all these functions in the handler after we define the functions in the other modules

Lets inspect out data_transformation module first

```python
def join_dfs(df, jh, col='Date'):
    ''' Takes in two dataframes and preforms a left join
    that happens on the Date column, change the col argument
    for a different column'''
    final = df.merge(jh, on=col, how='left')
    return final


def change_to_datetime(df, col='Date'):
    ''' Changes a column in a dataframe to type datetime'''
    df[col] = df[col].astype('datetime64[ns]')
    return

def rename_cols(df, names_dict):
    ''' Renames a group of columns in a df by passing a dict of old names and new names  '''
    df.rename(columns=names_dict, inplace=True)
    return


def clean_dataframe(jh, country, columns, col='Country/Region'):
    ''' Filters a dataframe column from entries that don't
    match a specific condition and removes needless columns,
    used for cleaning the John Hopkins data'''
    jh.drop(jh[jh[col] != country].index, inplace=True)
    jh.drop(columns, axis=1, inplace=True)
    return
```

Our end dataframe should combine the case counts and deaths in the first data source to the recoveries column in the second data source. the second data source also contains data from many countries, for the sake of our case we only want USA data. and so we define functions for dropping the columns and data we don't need, unifying the naming structure and data types in both dataframes, and finally join them together.


And now, let's look at our database module


```python
import psycopg2 #postgres client
import boto3
import json

sm = boto3.client('secretsmanager')

secret = sm.get_secret_value(SecretId='postgres-credentials')
secrets_dict = json.loads(secret['SecretString'])

def connect():
    ''' Instantiates a connection to the db'''
    conn = psycopg2.connect(
        host=secrets_dict['host'],
        database='postgres',
        user=secrets_dict['username'],
        password=secrets_dict['password'])
    return conn

def instantiate_db():
    conn = connect()
    cursor = conn.cursor()
    try:
        cursor.execute('SELECT * FROM covid19 LIMIT 1;')
        return
    except UndefinedTable:
        print('creating table')
        cursor.execute(
            '''CREATE TABLE covid19 (
                date TIMESTAMP PRIMARY KEY,
                cases NUMERIC,
                deaths NUMERIC,
                recovered NUMERIC
            );

            INSERT INTO covid19 (date, cases, deaths, recovered) VALUES ('2019-01-01', 0, 0, 0);'''
        )
        conn.commit()
        cursor.close()
        conn.close()
    
 
def get_latest_date():
    ''' Returns the current ltest record in the db'''
    conn = connect()
    cursor = conn.cursor()
    cursor.execute('SELECT * FROM covid19 ORDER BY date DESC LIMIT 1;')
    results = cursor.fetchall()
    cursor.close()
    conn.close()
    return results[0][0]
    

def load_to_db(row):
    ''' Insert new rows to the db'''
    conn = connect()
    cursor = conn.cursor()
    cursor.execute(
        'INSERT INTO covid19 (date, cases, deaths, recovered) VALUES (%s, %s, %s, %s)', (row['Date'], row['cases'], row['deaths'], row['Recovered'])
        )
    conn.commit()
    cursor.close()
    conn.close()
```

Firstly, we don't have the database credentials hardcoded or even known ahead of deploying. we use AWS's secrets manager to create a secure password for us and retrieve it from the API at runtime. 
This approach is extremely better for security as the credentials are never visible to anyone and only available during processing


After having retrieved the credentials we instantiate a connection using psycopg2, the python API for PostgreSQL

```python
def connect():
    ''' Instantiates a connection to the db'''
    conn = psycopg2.connect(
        host=secrets_dict['host'],
        database='postgres',
        user=secrets_dict['username'],
        password=secrets_dict['password'])
    return conn
```

We return the connection object as its essential for any database operation, this function is a dependency for all other functions in the module.

Second, we have a function for creating our table. this function will first attempt to query for the existence of the table and return a random record from it. if it fails to do so due to `UndefinedTable` it will create the table and insert a dummy record that dates back to before the data in our sources. this is important for the `get_latest_date()` function to work properly, because if it retrieves this dummy record, initial loading is performed.

lastly, we have the actual function that loads a record into the database. this function expects a dataframe row as input.

The justification behind this approach is that dataframe operations that involve iteration are slow according to this [Stackoverflow question](https://stackoverflow.com/questions/16476924/how-to-iterate-over-rows-in-a-dataframe-in-pandas).
One of the alternatives suggested is to use `df.apply(your_function)`, which is how we formulate our approach.


And now, we look at the actual handler code that executes at invocation of the function


```python
def lambda_handler(event, context):
    try:
        download_file('https://raw.githubusercontent.com/nytimes/covid-19-data/master/us.csv', 'data.csv')
        download_file('https://raw.githubusercontent.com/datasets/covid-19/master/data/time-series-19-covid-combined.csv', 'john_hopkins.csv')
        df = load_data('data/data.csv')
        jh = load_data('data/john_hopkins.csv')
        rename_cols(df, {'date':'Date'}) #unify naming method
        clean_dataframe(jh, 'US', ['Deaths', 'Country/Region' ,'Province/State',  'Confirmed']) # remove needless columns and non US data
        change_to_datetime(df)
        change_to_datetime(jh)
        final = join_dfs(df, jh)
        instantiate_db() # create the table if it doesnt exist
        data_sent = send_db(final) 
        res = post_to_sns('Job ran successfully, Updated {} records in the database'.format(data_sent)) # report if successful
    except Exception as e:
        res = post_to_sns('Job failed with error: {}'.format(e)) # report errors
```

## Infrastructure as Code

For this project we use the AWS CDK, a framework by AWS for defining your infrastructure as actual code in your programming language of choice, which is worth saying that it is a true wonder of engineering and i had so much fun using it.
However it is worth saying that navigating the [docs](https://docs.aws.amazon.com/cdk/api/v2/python/index.html) was extremely tedious and some of the processes are not straight forward. for the sake of keeping the project entirely in Python we deploy CDK in python flavor.

Resources in a CDK stack are defined with the following parameters:

- the construct that will use the resource

- an identifier

- any properties specific to the resource 

Learn more about the CDK in this [workshop](https://cdkworkshop.com/)

Let's take a look at our CDK stack

```python
class CdkStack(Stack):

    def __init__(self, scope: Construct, construct_id: str, **kwargs) -> None:
        super().__init__(scope, construct_id, **kwargs)

        # Define Lambda layers
        pandas = lambda_.LayerVersion.from_layer_version_attributes(self, 'Pandas',
            layer_version_arn="arn:aws:lambda:us-east-1:770693421928:layer:Klayers-p39-pandas:4")

        requests = lambda_.LayerVersion.from_layer_version_attributes(self, 'Requests',
            layer_version_arn="arn:aws:lambda:us-east-1:770693421928:layer:Klayers-p39-requests-html:4")

        psycopg = lambda_.LayerVersion.from_layer_version_attributes(self, 'Psycopg',
            layer_version_arn="arn:aws:lambda:us-east-1:770693421928:layer:Klayers-p39-psycopg2-binary:1")
```

Before anything, Huge shoutout to Keith Rozario for providing us with this awesome [Repo](https://github.com/keithrozario/Klayers).
where he defined layers for lambda functions to add libraries that doesn't exist by default in the lambda runtime without you having to create them yourself. I used his repo to add the requests, pandas and psycopg2 libraries to the project.

Next up, let's define our database..

```python
# instantiate DB credentials using secrets manager
        db_secrets = rds.DatabaseSecret(self, 'postgres-secret',
                    username='postgres',
                    secret_name='postgres-credentials'
                    )

        # Create the database
        db = rds.DatabaseInstance(self, "db",
                    engine=rds.DatabaseInstanceEngine.postgres(version=rds.PostgresEngineVersion.VER_13_4),
                    instance_type=ec2.InstanceType.of(ec2.InstanceClass.BURSTABLE3, ec2.InstanceSize.MICRO),
                    credentials=rds.Credentials.from_secret(db_secrets),
                    vpc=ec2.Vpc(self, 'vpc'),
                    vpc_subnets=ec2.SubnetSelection(
                        subnet_type=ec2.SubnetType.PUBLIC
                    )
                    )

db.connections.allow_default_port_from_any_ipv4()
```
We instantiate the database secret using Secrets Manager and then create the database using this secret using the `rds.Credentials.from_secret(db_secrets)` method, we also launch this DB in a public subnet for easier reach from AWS Lambda.
However in a production setting and in case the database is going to hold sensitive data its best to launch it in a private subnet and configure Lambda to work in your private network, check this [article](https://docs.aws.amazon.com/lambda/latest/dg/configuration-vpc.html#:~:text=When%20you%20connect%20a%20function,enabled%20function%20in%20an%20account.) for more details.


And now, let's create our S3 bucket and SNS topic. pretty straight forward

```python
# Create the bucket used to store the data
        s3_bucket = s3.Bucket(self, 'dataBucket')

        # Create the delivery topic
        topic = sns.Topic(self, 'deliveryTopic')
```

And defining our actual function with appropriate permissions..

```python
lambda_role = iam.Role(self, "lambda_role",
                    assumed_by=iam.ServicePrincipal("lambda.amazonaws.com")
                    )
        lambda_role.add_to_policy(iam.PolicyStatement(
                    resources=["*"],
                    actions=[
                            "s3:GetObject",
                            "s3:PutObject",
                            "sns:Publish",
                            "secretsmanager:GetSecretValue"]
                    ))
function = lambda_.Function(self, "Serverless-ETL",
                    runtime=lambda_.Runtime.PYTHON_3_9,
                    code=lambda_.Code.from_asset("./code"),
                    handler="lambda_function.lambda_handler",
                    layers=[pandas, requests, psycopg],
                    timeout=Duration.minutes(5),
                    role=lambda_role,
                    memory_size=512,
                    environment={
                        'S3_BUCKET':s3_bucket.bucket_name,
                        'SNS_TOPIC': topic.topic_arn
                    }
                    )
# allow our function to post to sns
topic.grant_publish(function)

```

As we see here, the bucket name and SNS topic ARN are passed to the function on deployment through its environment variables. and we load the lambda code saved in a folder in the directory called "code" using the method `lambda_.Code.from_asset("./code")`. we also configure the function with 512MB of memory, this is because this function instantiates several clients and they take up alot of memory to work with.

Lastly, we define the Eventbridge rule to trigger our function

```python
        # Create the event rule and schedule
        rule = events.Rule(self, "Rule",
                    schedule=events.Schedule.expression('cron(0 0 * * ? *)'),
                    )        
        rule.add_target(targets.LambdaFunction(function))
```

## Continuous Delivery

The beauty of using the AWS CDK as your tool of choice for deploying is its simplicity in use and maintaining of infrastructure state using Cloudformation changesets. You can read more about them [here](https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/using-cfn-updating-stacks-changesets.html)

The only commands we need to launch our infrastructure are `cdk diff` to detect changes and `cdk deploy` to commit changes. we can wrap these commands in a nice and easy Github actions workflow as after having configured your AWS access keys in Github. here's a [link](https://github.com/aws-actions/configure-aws-credentials) explaining how its done.

```yaml

# run only on commits to main
on:
  push:
    branches: [ "main" ]
  pull_request:
    branches: [ "main" ]

  workflow_dispatch:


jobs:
  aws-cdk:
    runs-on: ubuntu-latest
    steps:
    
      - name: Check out repository code
        uses: actions/checkout@v3
    
      - name: cdk diff
        uses: youyo/aws-cdk-github-actions@v2
        with:
          cdk_version: 2.28.1
          cdk_subcommand: 'diff'
          cdk_stack: CdkStack
          actions_comment: false
        env:
          AWS_ACCESS_KEY_ID: ${{ secrets.AWS_ACCESS_KEY_ID }}
          AWS_SECRET_ACCESS_KEY: ${{ secrets.AWS_SECRET_ACCESS_KEY }}
          AWS_DEFAULT_REGION: 'us-east-1'
          
          
      - name: cdk deploy
        uses: youyo/aws-cdk-github-actions@v2.1.1
        with:
          cdk_version: 2.28.1
          cdk_subcommand: 'deploy'
          cdk_stack: 'CdkStack'
          cdk_args: '--require-approval never'
          actions_comment: false
        env:
          AWS_ACCESS_KEY_ID: ${{ secrets.AWS_ACCESS_KEY_ID }}
          AWS_SECRET_ACCESS_KEY: ${{ secrets.AWS_SECRET_ACCESS_KEY }}
          AWS_DEFAULT_REGION: 'us-east-1

```

This workflow uses actions from the actions marketplace, find out more [here](https://github.com/marketplace/actions/aws-cdk-github-actions). For every new commit to change our code for the function or the infrastructure, the CDK CLI will run and show us what changed and will deploy these changes to your account.

In the end i hooked up Amazon Quicksight to my RDS database to extract insights on the data. RDS integrates with Quicksight very seamlessly
![BI reporting](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/3rundyf62akezrl97hbg.png)


## Conclusion

Modern cloud infrastructure is being pushed to entirely new limits and possibilities and its only up to your imagination how you will use it. I have enjoyed a lot and learned so much while creating this. all the code for this article can be found in this [Repo](https://github.com/aerosouund/serverless-etl-cdk)