DEV Community: Sean O'Connor

🚨 Introducing a python package that helps with machine learning feature engineering during the pandemic! lockdowndates🔥

Sean O'Connor — Thu, 17 Mar 2022 12:58:10 +0000

Recently during the pandemic if you were training machine learning models with some sort of time element you would have been scrambling to try and reengineer these as I would be confident to say the pandemic made all your forecasts very far from the truth.

Consumer habits were completely changed and became so unexpected it was hard to forecast. Everyone was left in the dark as to whether we all would be thrown into a complete or partial lockdown.

As a machine learning engineer I had to completely change my approach to training machine learning models during the pandemic. It took a lot of tweaking. The hardest part was training models for consumers in different countries around the world, where restrictions were completely different. This is still a pain for some countries at the moment with restrictions coming and going.

So, I decided instead of trying to alleviate the damage of the pandemic in our models, why not embrace them? Why not embed them as features in our models for them to learn about? That's why I decided to created lockdowndates

lockdowndates* provides all current and past restrictions imposed by governments in over **100 countries worldwide during the pandemic! To get started:

pip3 install lockdowndates

(side note: currently we only support python3.8 and above, this will be changing soon to support 3.6>)

after installing we import:

from lockdowndates.core import LockdownDates

to get restrictions for a single country:

ld = LockdownDates("Aruba", "2022-01-01", "2022-01-08")
lockdown_dates = ld.dates()
lockdown_dates

	aruba_country_code	aruba_stay_at_home
timestamp
2022-01-01	ABW	2.0
2022-01-02	ABW	2.0
2022-01-03	ABW	2.0
2022-01-04	ABW	2.0
2022-01-05	ABW	2.0
2022-01-06	ABW	2.0
2022-01-07	ABW	2.0
2022-01-08	ABW	2.0

to get restrictions for multiple countries:

ld2 = LockdownDates(["Canada", "Denmark"], "2022-01-01", "2022-01-08")
lockdown_dates = ld2.dates()
lockdown_dates

	canada_country_code	denmark_country_code	canada_stay_at_home	denmark_stay_at_home
timestamp
2022-01-01	CAN	DNK	1.0	0.0
2022-01-02	CAN	DNK	1.0	0.0
2022-01-03	CAN	DNK	1.0	0.0
2022-01-04	CAN	DNK	1.0	0.0
2022-01-05	CAN	DNK	1.0	0.0
2022-01-06	CAN	DNK	1.0	0.0
2022-01-07	CAN	DNK	1.0	0.0
2022-01-08	CAN	DNK	1.0	0.0

the legend for stay_at_home are as follows:

NaN - No data available for that date.
1.0 - recommend not leaving house.
2.0 - require not leaving house with exceptions for daily exercise, grocery shopping, and 'essential' trips.
3.0 - require not leaving house with minimal exceptions (eg allowed to leave once a week, or only one person can leave at a time, etc.

lockdowndates contains up to date data thanks to oxford university and their open source data!

I will keep updating lockdowndates to contain more restrictions, including restrictions for vaccinated vs non-vaccinated, school restrictions, work from home restrictions and many more! Track our issues here.

So next time you feature engineering on your next machine learning project and have tabular data from the during the pandemic, consider lockdowndates to make your life easier and to improve your metrics!

Documentation: lockdowndates

Github Repo: lockdowndates

Spinning up an AWS Fargate service based on messages on a queue - using aws-cdk ☁️👀

Sean O'Connor — Tue, 04 Jan 2022 18:07:08 +0000

Until recently I had an ECS service that would listen constantly to an SQS queue waiting until there were any messages and then would process the message. This was a waste of resources and not cost-effective at all. So, I looked into alternative ways where I could simply spin up tasks only when messages were pushed to the SQS queue. This would mean my service would not lay idle all day and all night waiting.

Enter aws-cdk. Please bear in mind this is not a beginners tutorial but one that caused me a lot of headaches and iterations to get right. Hopefully, this will help at least one person/team and save them from weeks of pain.

Prerequisites:

Experience with aws-cdk.
Experience with python.
Experience with aws-sdk.

By the end of the tutorial, you will have a Fargate service that spins up tasks when messages are pushed to an SQS queue. I will also show an alternative way to get the service to spin down these tasks when there are no messages on the queue due to some issues around current library implementations.

Quickstart

Install the kit with npm install -g aws-cdk

To check it has been installed cdk --version

To initiate a project:

mkdir fargate-cdk
cd fargate-cdk
cdk init sample-app --language=python

Imports

To begin with, import every service under the sun that aws-cdk provides.

from aws_cdk import (
    core as cdk,
    aws_ec2 as ec2,
    aws_ecs as ecs,
    aws_ecs_patterns as ecs_patterns,
    aws_sqs as sqs,
    aws_ecr as ecr,
    aws_iam as iam,
    aws_ssm as ssm,
    aws_kms as kms,
    aws_autoscaling as autoscaling,
    aws_applicationautoscaling as app_autoscaling,
    aws_cloudwatch as cw,
)
from aws_cdk.aws_cloudwatch_actions import ApplicationScalingAction
from aws_cdk.core import Duration

Initialise and fetch variables

Once you have used the aws-cdk command to initialise your project, navigate to sample-app.py file and you should see a class created by aws-cdk that inherits from cdk.Stack. We shall be placing all our code under the __init__ method for this tutorial.

Throughout the code, you will need some basic variables that will be used throughout so let's define those first.

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

        image_tag = "" # tag of the repo within ecr

        cluster_name = "" # cluster group name of containers

        security_group = "" # virtual firewall for your EC2 instances

        kms_key = "" # needed for decrypting

        queue_name = "" # name of sqs queue that will be created

        vpc_id = "" # id of the VPC in which to provision the cluster.
...

Using variables to fetch objects

We will now use the defined variables to fetch our vpc, cluster, kms key and to create our sqs queue.

...
        # fetches vpc using the id
        vpc = ec2.Vpc.from_lookup(self, "VPC", vpc_id=vpc_id)

        # fetches kms_key used for decryption using the arn of the key
        kms_key = kms.Key.from_key_arn(
            self, "EncryptionKey", f"arn:aws:kms:us-west-1:123456789:key/{kms_key}" # arn of kms key
        )

        # fetches cluster using defined objects and variables above
        cluster = ecs.Cluster.from_cluster_attributes(
            self,
            "Cluster",
            vpc=vpc,
            cluster_name=cluster_name,
            security_groups=[
                ec2.SecurityGroup.from_security_group_id(
                    self, "SecurityGroup", security_group_id=security_group
                )
            ],
        )

        # creates a fifo queue
        queue = sqs.Queue(
            self,
            "SampleQueue",
            content_based_deduplication=True,
            fifo=True,
            queue_name=queue_name, # must contain .fifo at end of name
            receive_message_wait_time=Duration.seconds(20),
            retention_period=Duration.seconds(345600),
            visibility_timeout=Duration.seconds(43200),
        )
...

Fetching secrets from Systems Manager Parameter Store and defining environment variables

THIS is a very key step.

We must define any parameters that are stored within the systems manager parameter store and that will be used within your application.

...

        database_name = ssm.StringParameter.from_secure_string_parameter_attributes(
            self,
            "DATABASE_NAME",
            version=1,
            parameter_name="", # name of parameter within ssm
            simple_name=True, # if using name of parameter and not arn or any other identifier
            encryption_key=kms_key, # needed if encryption is turned on within parameter store. Normally needed when using simple_name
        )
...

Then create a dictionary of these secrets. Any variables that your app fetches from the local environment must be defined too.

...
        secrets = {
            "DATABASE_NAME": ecs.Secret.from_ssm_parameter(database_name),
        }

        environments = {
            "ENV": environment,
        }
...

Creating the Queue Processing Fargate Service

This is the most important part. It creates the Fargate service and uses everything we have defined up until now.
This service will listen to the queue created and spin up tasks when there are messages on the queue.

...
        fargate_service = ecs_patterns.QueueProcessingFargateService(
            self,
            "SampleAppService",
            cpu=4096, # set the amount of cpu you wish your tasks to have
            memory_limit_mib=8192, # set the amount of memory you with your task to have
            image=ecs.ContainerImage.from_ecr_repository(
                repository=ecr.Repository.from_repository_name(
                    self,
                    "SampleAppRepository",
                    repository_name="",
                ), # fetch repo name from ecr
                tag=image_tag,
            ), # fetch image from a certain ecr repo
            min_scaling_capacity=0, # min number of tasks
            max_scaling_capacity=1, # the max number of tasks a service can spin up
            cluster=cluster, # cluster for the service
            queue=queue, # queue the service shall be listening to.
            scaling_steps=[
                {"upper": 0, "change": 0},
                {"lower": 1, "change": +1},
            ], # this defines how the service shall autoscale
            secrets=secrets, # secret parameters from ssm
            environment=environments, # any additional env variables needed
            visibility_timeout=cdk.Duration.hours(12) # how long the task will take the item off the queue for
        )
...

DO NOT use the variable desired_task_count as it is deprecated and will not have an effect on the service at all.

Key points:

Make sure you set a visbility_timeout or when you try to delete the message from the queue after 30 seconds it will fail due to the message id timing out. The default timeout is 30 seconds.
It is important to set the upper scaling step to {"upper": 0, "change": 0}. This is because when the task takes the last message off the queue we do not want any change to occur to the service. If we set the change to be -1 the service will try to tear down the task whilst it is processing the last message. So it is key that when there are 0 messages on the queue we do not kill our task. We will solve this issue with CPU alarms later. See this link for an ongoing Github issue.

Adding policies to the service

Now we have set up our service we need to add policies to it so our task application can take action on other aws services.

...
        kms_policy = iam.PolicyStatement(
            actions=[
                "kms:Decrypt",
            ],
            effect=iam.Effect.ALLOW,
            resources=[
                "", # the arn of the resource
            ])

fargate_service.task_definition.add_to_task_role_policy(kms_policy)
...

Deployment - Stage 1

Right! Now we have a Fargate service that scales when messages are on the SQS queue. This is stage 1 of deployment and currently, our service can only spin up, and not tear down tasks.

To allow our tasks to tear down, we need to deploy the code we have produced above. I'll explain why in the next section.

To deploy: cdk deploy

Scaling down task

Now we have deployed the first step, how do we go about getting our service to tear down our task when there are no messages?

Early I mentioned we could not tear down based on messages on the queue due to the fact it would kill a task mid processing. So to scale down we will focus on the CPU usage of the task. If our tasks CPU usage is below a threshold for a length of time we will scale our tasks down to 0. This way we get around our tasks being killed whilst processing as CPU usage will be high when the task is in use.

So now add below the Fargate service and before the policies:

...
        # initialise the fargate service CPU usage metric to average over 3 minutes
        fargate_service_cpu_metric =         fargate_service.service.metric_cpu_utilization(
            period=Duration.minutes(3),
            statistic="Average"
        )

        # add an alarm to our fargate service CPU usage
        scale_in_init = fargate_service_cpu_metric.create_alarm(
            self,
            'SampleApp-ZeroCPUAlarm',
            alarm_description="For when sample app is idle, scale service to 0",
            alarm_name=alarm_name,
            evaluation_periods=1,
            threshold=0.01, # set threshold of cpu usage.
            actions_enabled=True,

        # create comparison operator so that we compare our cpu usage with our threshold. We want it less than or equal to the threshold             comparison_operator=cw.ComparisonOperator.LESS_THAN_OR_EQUAL_TO_THRESHOLD,
            datapoints_to_alarm=1,
            statistic="Average",
        )

        # define our auto scaling target for our fargate service 
        scalable_target = app_autoscaling.ScalableTarget.from_scalable_target_id(
            self,
            'SampleApp-ScalableTarget',
            scalable_target_id=f'service/{fargate_service.cluster.cluster_name}/{fargate_service.service.service_name}|ecs:service:DesiredCount|ecs',
        )

        # define the action taken on our scaling target
        scaling_action = app_autoscaling.StepScalingAction(
            self,
            "scaleToZero",
            scaling_target=scalable_target,
            adjustment_type=autoscaling.AdjustmentType.EXACT_CAPACITY,
        )

        # create the adjustment made by our action
        scaling_action.add_adjustment(adjustment=0, upper_bound=0)

        # finally add our alarm action to our fargate alarm
        scale_in_init.add_alarm_action(ApplicationScalingAction(scaling_action))
...

Deployment - Stage 2

Now, we have to redeploy our stack so our alarm is activated on our Fargate service.

So now, when our tasks are inactive and CPU usage is below the 0.01 threshold. Our Fargate service will spin down our tasks.

Conclusion

Hopefully, that will save someone a huge amount of pain. I am definitely a fan of using aws-cdk instead of cloudformation as it allows you to deal with traditional programming languages instead of pure YAML.

I will definitely be iterating on this code to remove the need for two-stage deployments by taking advantage of pipelines. I will update this tutorial as I learn more.

Fetching and reading files from S3 using Go 🔥👀

Sean O'Connor — Wed, 26 May 2021 15:26:11 +0000

Trying to figure out how to do simple tasks using the AWS SDK for particular services can be difficult given that sometimes the AWS documentation is limited and gives you the bare minimum. Today I'll show you how to fetch and read particular files from S3 using Go. This tutorial collates many hours of research into what should be a simple problem.

Prerequisites include:

Go installed / previous experience with Go.
AWS-SDK set up / previous development with AWS-SDK.

Basic imports

import (
    "encoding/json"
    "fmt"
    "io/ioutil"
    "log"

    "github.com/aws/aws-lambda-go/lambda"
    "github.com/aws/aws-sdk-go/aws"
    "github.com/aws/aws-sdk-go/aws/session"
    "github.com/aws/aws-sdk-go/service/s3"
)

Defining global variables and structs.

Start off by defining some basic structs and global variables.


type S3Bucket struct {
    Bucket string `json:"bucket"`
    Key    string `json:"key"`
}

type Metrics struct {
    RMSE         string      `json:"rmse"`
    MAE        string      `json:"mae"`
    MAPE        string      `json:"mape"`
}

var pageNum int = 0
var s3Buckets []S3Bucket
var finalMetrics []Metrics
var sess *session.Session

Initiating a session.

Firstly we initialise a session that the SDK uses to load credentials from the shared credentials file ~/.aws/credentials, and create a new Amazon S3 service client.

sess, err := session.NewSession(&aws.Config{
        Region: aws.String(conf.AWS_REGION),
    })

if err != nil {
    exitErrorf("Unable to create a new session %v", err)
}

Listing items in a bucket with pagination.

The AWS docs only give an example of accessing a bucket's files using ListObjectsV2 function. Now the problem I encountered with this function it does not allow us to apply our own custom function to the results in order for us to filter them even more. Another problem is it returns (up to 1,000) of the objects in a bucket with each request. This includes sub-paths to the files you wish to read.

ListObjectsV2 lists all objects in our S3 bucket tree, even objects that do not contain files. If I want to target certain objects we have to apply a function. So, instead we'll use ListObjectsV2Pages. ListObjectsV2Pages iterates over the pages of a ListObjectsV2 operation, calling the function with the response data for each page. To stop iterating, we return false.

As shown below I wish to target only the .json files in the page and append them to an s3Bucket slice. This part is important as it will allow us to know the location of each file so we can then access the contents!

We pass our main bucket name as S3_BUCKET and our object path if there is one into S3_PREFIX.

svc := s3.New(sess)
err = svc.ListObjectsV2Pages(&s3.ListObjectsV2Input{Bucket: aws.String(S3_BUCKET), Prefix: aws.String(S3_PREFIX)},
    func(page *s3.ListObjectsV2Output, lastPage bool) bool {
        pageNum++
        for _, item := range page.Contents {
            if strings.Contains(*item.Key, "json") {
                s3Buckets = append(s3Buckets, S3Bucket{Bucket: conf.S3_BUCKET, Key: *item.Key})
            }
        }
        return pageNum < 100
    })

if err != nil {
    exitErrorf("Unable to list items in bucket %q, %v", conf.S3_BUCKET, err)
}

Accessing the object contents.

Using the s3buckets slice, we will access the Bucket and Key from the struct and request the 'Object' information (or in other words the file) and then fetch the object based on the object information.

for _, item := range s3Buckets {
    requestInput := &s3.GetObjectInput{
        Bucket: aws.String(item.Bucket),
        Key:    aws.String(item.Key),
    }

    result, err := svc.GetObject(requestInput)
    if err != nil {
        log.Print(err)
    }

Reading the contents into slice

The JSON file 'result' is read with the ioutil.Readall()function, which returns a byte slice that is decoded into the Metrics struct instance using the json.Unmarshal() function.

The best tutorial I have found regarding reading JSON into a struct is this one: Parsing JSON

    defer result.Body.Close()
    body, err := ioutil.ReadAll(result.Body)
    if err != nil {
        log.Print(err)
    }

    bodyString := fmt.Sprintf("%s", body)
    var metrics Metrics
    err = json.Unmarshal([]byte(bodyString), &metrics)

    if err != nil {
        fmt.Println("twas an error")
    }

    finalMetrics = append(finalMetrics, metrics)

}

And that's it! You have now fetched JSON files from a certain bucket and parsed the results into a struct. In my opinion, especially in machine learning, fetching the contents of an S3 file is hugely important as engineers we are constantly wanting to see and compare for example past models' performance or fetching additional data features to append to our models.