Tutorial: Creating a Data Entry Microservice with Go and CDK

Transfer and sanitize data from documents to DynamoDB

Photo by Mika Baumeister on Unsplash

Introduction

Manual data processes lead to poor data quality, which costs organizations an average of $12.9 million to $15 million USD annually according to Gartner. Beyond the financial impact, manual processes introduce security vulnerabilities and create bottlenecks that frustrate users. I’ve built variations of this batch import pattern in JavaScript and Python over the years, but this implementation uses the Go programming language. According to benchmark tests, Go is a more versatile and efficient language (especially for web development) compared to the more popular JavaScript and Python. In production environments processing millions of rows daily, Go’s CSV parsing is 3–4x faster than Python and uses significantly less memory than Node.js, making it the pragmatic choice for this workload. This article walks through building a production-ready CSV ingestion pipeline on AWS that I’ve deployed for enterprise clients. We’ll cover not just the implementation, but the architectural decisions, cost trade-offs, and operational lessons that distinguish a tutorial project from a system that runs reliably at scale. So let’s stop costing our companies millions and get it together. The tutorial is broken down into these parts, I recommend following them in order:

1. Setting Up the CDK environment
2. Designing the Microservice Architecture
3. Implementing the Microservice Components
4. Deploying
5. Testing
6. Optimizing

Setting Up the CDK Environment

Assuming you have already created an AWS account and have admin access to set up AWS IAM roles and permissions, we can start with the CDK setup. The meaty part of this involves setting up the CDK environment using your terminal and choosing the appropriate AWS services for the microservice.

What is CDK?

Long story short it’s a way to write recipes for creating AWS infrastructure using regular programming languages like Python, Java, JavaScript, Go and a few others. You may have seen me using other mechanisms such as Cloudformation and Amplify in the past to create AWS infrastructure. Those are all fine and well; they use YAML or JSON templates to set up what I need. However, CDK really gives me a leg up for a few reasons:

• Reduced boilerplate
• Multiple resources of the same configuration can be created using loops
• Construct libraries exist for almost any pattern you can imagine
• Type safety and IDE support
• Easier version control and CI/CD compared to Amplify
• CDK code can be exported as Cloudformation

CLI commands for setup

Installation is pretty easy:

$ npm install -g aws-cdk

Then set up a directory to act as your workspace

$ mkdir project-cdk && cd project-cdk

Initialize your CDK

$ cdk init app --language go
$ export CDK_DEFAULT_ACCOUNT="XXXXXXXXXXXX"
$ export CDK_DEFAULT_REGION="us-east-1" # or whatever you prefer

Then wait for it to initialize the CDK workspace. Once that’s done you’ll also need to install Go programming language. At the time of writing this tutorial I am on the latest release of Go which is 1.23.5. On Mac you can verify it is installed with:

 $ which go
/usr/local/go/bin/go

Then install the CDK dependencies. Use go get to add the required CDK packages we’ll need for this tutorial to your project:

$ go get github.com/aws/aws-cdk-go/awscdk/v2

You can optionally run go mod tidy to tidy up the go.mod file and download any missing dependencies at any time.

AWS Services We’ll Be Working With (in case you’re not familiar)

• DynamoDB: A NoSQL Database
• Lambda: Serverless Compute Service
• SQS: Messaging Queue Service
• S3: Object Storage
• CloudFormation: Infrastructure Management Tool

Designing the Microservice Architecture

Before we start designing the architecture and data flow between AWS services it’s important to understand the purpose and advantages of microservices. Just like we have components for each logical entity of the frontend in React.js, microservices are a way to accomplish separating the backend into independently replaceable and upgradeable parts. The benefits of this are wide and variable. Reduced blast radius is the most common example. Say for example you have a monolithic backend system with authentication, inventory, payment, vendor management, and business reporting all rolled into one. You push out a code change that contains a memory leak in the business reporting that takes down the entire site. If that part of the backend were to have its own database, API, etc. then the rest of the site would still be running while you fix it. Furthermore, if you wanted to use a different runtime, host, or database management system for each microservice in the backend then that is also an advantage it has over monoliths. Something like the data entry service we will build in this tutorial could make up part of a larger backend architecture. For that reason we have to pay attention to scalability and performance.

How this is going to look

Cloud Architecture Design

Designing the architecture for any microservice starts with defining what you want your end product to be and then working your way back to the starting point. In this case our starting point is going to be the data stored in the database of our choosing. To get it there, some process will have to write it programmatically. In the absence of hard latency requirements and to keep costs down, we’ll rely on AWS Lambda Functions for this tutorial. In a real-world scenario Fargate would be a better choice. Always consult the latest AWS documentation when deciding on which compute service to use based on your needs.

Courtesy of AWS

The data we want to write to the database would be one batch of rows, which corresponds to rows in our CSV file. For this reason our Lambda function will accept a batch of rows from the CSV file and output a BatchWriteItem command for the DB. We can use SQS to pass rows from the CSV file to this function. The queue can hold many batches that were processed from the file while waiting for batches ahead of them to be written. We can feed this SQS queue with another Lambda function that can process the batches and send them out in chunks of a predetermined size (say 10 batches at a time for example). That way we can process multiple batches in parallel. More importantly we also need to potentially process multiple files uploads in parallel. We’ll have another Lambda function + SQS queue that processes the file uploads and batches them in a predetermined size (say no more than 3 files at a time).

The data will flow in order of:

• File uploaded to bucket →
• File queued →
• File processed →
• Batch of rows read →
• Individual batch queued →
• Individual batch processed →
• DB Records queued →
• DB Records written to the DB

Each queue will also contain a dead letter counterpart to process any file, batch, row, or DB record that fails to be processed. You can then return these to the end user to retry uploading them. Decoupling each stage in the process like this allows for each stage to scale independently. You can tweak the size of a queue, batch, concurrent functions, and other resources to meet your needs. As always, choosing the right AWS service for each component will make you worth your weight in gold. This applies not only for performance reasons but also for cost optimization.

Photo by Tania Melnyczuk on Unsplash

Implementing the Microservice Components

Let’s start off by thinking about the parameters we will need for the entire microservice:

• Stage: Is this a sandbox, gamma, beta, or prod environment (hopefully you also have different AWS accounts for these!)
• Service Name: Pretty self-explanatory, what do we want to call this micro-service. Feel free to go as wild as you want with this one
• Project Tag: A unique name for your project
• Others: DynamoDB params (Read/Write Capacity Units), SQS params (Message Retention Period, Deadletter Queue Visibility Timeout, Batch Queue Visibility Timeout)

These parameters are going to be added in the csv_microservice.go file →

package main

import (
 "os"
 "fmt"
 "github.com/aws/aws-cdk-go/awscdk/v2"
 "github.com/aws/constructs-go/constructs/v10"
 "github.com/aws/jsii-runtime-go"
)

type CsvMicroserviceStackProps struct {
 awscdk.StackProps
 projectTag                       string
}

func NewCsvMicroserviceStack(scope constructs.Construct, id string, props *CsvMicroserviceStackProps) awscdk.Stack {
 var sprops awscdk.StackProps
 if props != nil {
  sprops = props.StackProps
 }
 stack := awscdk.NewStack(scope, &id, &sprops)

 return stack
}

func main() {
 defer jsii.Close()

 app := awscdk.NewApp(nil)

 NewCsvMicroserviceStack(app, "CsvMicroserviceStack", &CsvMicroserviceStackProps{
  awscdk.StackProps{
   Env: env(),
  },
  projectTag,
 })

 app.Synth(nil)
}

func env() *awscdk.Environment {
 return &awscdk.Environment{
  Account: jsii.String(os.Getenv("CDK_DEFAULT_ACCOUNT")),
  Region:  jsii.String(os.Getenv("CDK_DEFAULT_REGION")),
 }
}

For reference, the inherent environment variables built in to CDK code are:

• CDK_DEFAULT_ACCOUNTThe AWS account ID used for deployment
• CDK_DEFAULT_REGIONThe AWS region where the stack is being deployed
• CDK_CONTEXT_JSONA JSON-encoded string containing CDK context values
• CDK_OUTDIRThe output directory where CDK generates synthesized CloudFormation templates (cdk.out)
• CDK_CLI_ASM_VERSIONThe version of the AWS CDK CLI being used

In CDK you deal with Stacks and Resources. The TL;DR explanation of those two concepts is:

• Resources are any single piece of cloud infrastructure. Think an EC2 Instance (i.e. a cloud server), IAM Role, or S3 Bucket
• Stacks are a way to group multiple Resources together

I like to change the default name of the stack and even split infrastructure into multiple stacks if necessary. This helps to isolate issues and deploy or rollback parts of the application independently.

Integrating AWS S3 for file storage

The next thing we are going to do create our first Resource in the Stack. As you can see in the diagram above we will need an S3 Bucket. However, that is the simple description of what gets created. Behind the scenes we will need additional resources to make the Bucket usable. Starting with the S3 bucket we can create the following:

package main

import (
 "os"
 "fmt"
 ....
 "github.com/aws/aws-cdk-go/awscdk/v2/awss3"
)

type CsvMicroserviceStackProps struct {
 awscdk.StackProps
 projectTag                       string
}

func NewCsvMicroserviceStack(scope constructs.Construct, id string, props *CsvMicroserviceStackProps) awscdk.Stack {

  ....
 
 // Our paramters mentioned earlier
 stageParameter := awscdk.NewCfnParameter(stack, jsii.String("StageParam"), &awscdk.CfnParameterProps{
  Type:        jsii.String("String"),
  Description: jsii.String("Is this a sandbox, gamma, beta, or prod environment?"),
  Default:     jsii.String("sandbox"),
 })
 
 serviceNameParameter := awscdk.NewCfnParameter(stack, jsii.String("ServiceNameParam"), &awscdk.CfnParameterProps{
  Type:        jsii.String("String"),
  Description: jsii.String("What do we want to call this microservice?"),
  Default:     jsii.String("microservice1"),
 })

projectParameter := awscdk.NewCfnParameter(stack, jsii.String("ProjectParam"), &awscdk.CfnParameterProps{
  Type:        jsii.String("String"),
  Description: jsii.String("A unique name for your project"),
  Default:     jsii.String("deathramps"),
 })

 // Create the S3 bucket
 awss3.NewBucket(stack, jsii.String("S3Bucket"), &awss3.BucketProps{
  BucketName: jsii.String(fmt.Sprintf("%s-%s-%s", *projectParameter.ValueAsString(), *serviceNameParameter.ValueAsString(), *stageParameter.ValueAsString())),
 })

 return stack
}

func main() {
 ....
}

func env() *awscdk.Environment {
 ....
}

Here is a single bucket with a provided name. Later on we’ll need to add permissions for the bucket to call a Lambda function each time a CSV file gets uploaded. But for now, with this much code we can get started deploying and validating in the cloud. If you’re using a VPN you might need to pause it momentarily for this next part. From the project directory in your CLI run:

$ cdk synth

It should give you some output like this:

Parameters:
  StageParam:
    Type: String
    Default: sandbox
    Description: Is this a sandbox, gamma, beta, or prod environment?
  ServiceNameParam:
    Type: String
    Default: microservice1
    Description: What do we want to call this microservice?
  ProjectParam:
    Type: String
    Default: deathramps
    Description: A unique name for your project
  BootstrapVersion:
    Type: AWS::SSM::Parameter::Value<String>
    Default: /cdk-bootstrap/xxx000xxx/version
    Description: Version of the CDK Bootstrap resources in this environment, automatically retrieved from SSM Parameter Store. [cdk:skip]
Resources:
  S3Bucket00000000:
    Type: AWS::S3::Bucket
    Properties:
      BucketName:
        Fn::Join:
          - ""
          - - Ref: ProjectParam
            - "-"
            - Ref: ServiceNameParam
            - "-"
            - Ref: StageParam
    UpdateReplacePolicy: Retain
    DeletionPolicy: Retain
    Metadata:
      aws:cdk:path: CsvMicroserviceStack/S3Bucket/Resource

Before continuing with the rest of the cloud infrastructure, we can deploy what we already have and test that it’s working so far. Run cdk deploy from the command line:

cdk deployment

Dynamo DB Table

The next thing we want to add is a database and table within that database to hold the data we’re gonna pull from the spreadsheets. DynamoDB is a type of NoSQL database. When we define our database table we don’t need to be as verbose as we would when defining a traditional relational database table. Best of all we do not need to specify what our columns and their data types need to be, it’s all free form documents. In our case we want to add one table to hold a collection of users, with their ID, first name, and last name:

 -----------------------------
| id | first_name | last_name |
 -----------------------------
| 1  | Andrew     | Allison   |
| 2  | Jim        | Morrison  |

In our CDK code we can continue inside the code block of the NewCsvMicroserviceStack function below the S3 resources:

package main

import (

....

// New import
 "github.com/aws/aws-cdk-go/awscdk/v2/awsdynamodb" 
)

....

 // Create the DynamoDB Table
 tableName := "Users"
 awsdynamodb.NewTable(stack, jsii.String("DynamoDBTable"), &awsdynamodb.TableProps{
  TableName:    jsii.String(tableName),
  PartitionKey: &awsdynamodb.Attribute{
   Name:   jsii.String("id"),
   Type:   awsdynamodb.AttributeType_STRING,
  },
 });

 return stack
....

This definition is pretty simple and most times we would want to specify both a partition key and sort key as well as the read/write units. If you run cdk synth again you can see it added to the YAML output:

DynamoDBTable59784FC0:
    Type: AWS::DynamoDB::Table
    Properties:
      AttributeDefinitions:
        - AttributeName: id
          AttributeType: S
      KeySchema:
        - AttributeName: id
          KeyType: HASH
      ProvisionedThroughput:
        ReadCapacityUnits: 5
        WriteCapacityUnits: 5
      TableName: Users
    UpdateReplacePolicy: Retain
    DeletionPolicy: Retain

Optionally at this point you can deploy it to the cloud using cdk deploy or continue adding resources locally.

SQS Queues and Queue Policy

In case you need a refresher, we need to create two SQS queues (technically four since each queue will have an accompanying “dead letter queue”) for this microservice. One to process multiple, simultaneous file uploads and another to process the batches of rows of each file. We’ll begin with the queue for handling file uploads. The Dead Letter Queue will take care of all the messages that were sent and could not be processed. In this case if a CSV file is uploaded and for whatever reason it cannot be processed, a message will be sent to the DLQ (and later back to the end user) saying there was a problem:

package main

import (

....

// New import
  "github.com/aws/aws-cdk-go/awscdk/v2/awssqs"
)

....

 // Create the Batch Files SQS Queue
 batchFilesDLQueueName := fmt.Sprintf("%s-%s-%s-batch-files-dl-queue", *projectParameter.ValueAsString(), *serviceNameParameter.ValueAsString(), *stageParameter.ValueAsString())
 batchFilesDLQueue := awssqs.NewQueue(stack, jsii.String("BatchFilesDLQueue"), &awssqs.QueueProps{
  QueueName: jsii.String(batchFilesDLQueueName),
  RetentionPeriod: awscdk.Duration_Days(jsii.Number(1)),
 })

 batchFilesQueueName := fmt.Sprintf("%s-%s-%s-batch-files-queue", *projectParameter.ValueAsString(), *serviceNameParameter.ValueAsString(), *stageParameter.ValueAsString())
 awssqs.NewQueue(stack, jsii.String("BatchFilesQueue"), &awssqs.QueueProps{
  QueueName: jsii.String(batchFilesQueueName),
  VisibilityTimeout: awscdk.Duration_Seconds(jsii.Number(30)),
  DeadLetterQueue: &awssqs.DeadLetterQueue{
            MaxReceiveCount: jsii.Number(3),
            Queue: batchFilesDLQueue,
        },
 });

 return stack
....

The SQS queue has quite a few parameters:

• Visibility Timeout: The maximum time that it takes your application to process and delete a message from the queue
• DeadLetterQueue
• — MaxReceiveCount: Messages will be moved to DLQ after 3 failed processing attempts
• — Queue: The reference to the DLQ

For the sake of brevity I won’t cover what to do with the messages in the DLQ in this article but I will give some tips. For a CSV batch data upload process the dead letter strategy should focus on two things: alerting and recovery. For alerting, it’s possible to set up centralized AWS CloudWatch alarms that can let you as an admin know how the service is doing as a whole. For recovery, you can also send failed messages to an AWS SNS Topic that notifies whoever uploaded the botched CSV file to fix it and try again. This can work at both the file level and at the row level.

Next, we’ll follow up with another queue for handling the rows of a file:

....

// Create the Batch Rows SQS Queue
 batchRowsDLQueueName := fmt.Sprintf("%s-%s-%s-batch-rows-dl-queue", *projectParameter.ValueAsString(), *serviceNameParameter.ValueAsString(), *stageParameter.ValueAsString())
 batchRowsDLQueue := awssqs.NewQueue(stack, jsii.String("BatchRowsDLQueue"), &awssqs.QueueProps{
  QueueName:   jsii.String(batchRowsDLQueueName),
  RetentionPeriod: awscdk.Duration_Days(jsii.Number(1)),
 })

 batchRowsQueueName := fmt.Sprintf("%s-%s-%s-batch-rows-queue", *projectParameter.ValueAsString(), *serviceNameParameter.ValueAsString(), *stageParameter.ValueAsString())
 awssqs.NewQueue(stack, jsii.String("BatchRowsQueue"), &awssqs.QueueProps{
  QueueName:   jsii.String(batchRowsQueueName),
  VisibilityTimeout: awscdk.Duration_Seconds(jsii.Number(30)),
  DeadLetterQueue: &awssqs.DeadLetterQueue{
            MaxReceiveCount: jsii.Number(3),
            Queue: batchRowsDLQueue,
        },
 })

return stack
....

Run a quick cdk synth to make sure there are no typos or errors.

Lambda Functions

Up until this point the infrastructure we have added is static; resources are not communicating with each other and there is no computational work being done. AWS Lambda functions give us a chance to introduce code and do something with the data. When Lambda became generally available in 2015 it was a game-changer. Each AWS Lambda instance runs within a lightweight, isolated environment powered by Firecracker microVMs which are initialized with a runtime environment. All you have to do as a developer is write the code and set some parameters; you don’t have to worry about server management.

Get Andrew Allison’s stories in your inbox

Join Medium for free to get updates from this writer.

Remember me for faster sign in

To get started with Lambda we’ll add the relevant CDK code and a skeleton for the actual Lambda function source code. In our csv_microservice.go file:

package main

import (

....

// New import
  "github.com/aws/aws-cdk-go/awscdk/v2/awslambda"
)

....

 // Create File Upload Lambda Function
 awslambda.NewFunction(stack, jsii.String("FileUploadFunction"), &awslambda.FunctionProps{
  Description:       jsii.String("Handle a file upload"),
  Runtime:           awslambda.Runtime_PROVIDED_AL2023(),
  Handler:           jsii.String("main"),
  Code:              awslambda.Code_FromAsset(jsii.String(filepath.Join("functions", "file_uploaded")), nil),
  Environment:       &map[string]*string{
   "BUCKET_NAME":   jsii.String(bucketName),
   "QUEUE_URL": batchFilesQueue.QueueUrl(),
   "DL_QUEUE_URL": batchFilesDLQueue.QueueUrl(),
   "TABLE_NAME":   jsii.String(tableName),
  },
 })

return stack
....

Notice with this resource we are not giving it a name. The function name is optional and named Lambda functions are harder to update so it’s best to let the function get assigned a name by CDK / Cloudformation. There are 2 Environment variables for our function: the S3 bucket name and the SQS queue name. The function Code and Handler are referencing a local file path which has not yet been created, so lets do that now. In your CLI terminal, from the root project directory run:

$ mkdir functions && \ 
  cd functions && \
  mkdir file_uploaded && \
  cd file_uploaded && \
  go mod init file_uploaded && \
  go get github.com/aws/aws-lambda-go/lambda && \
  touch main.go

That will set you up with all the files you need for now. Going back to your IDE, you can open functions/file_uploaded/main.go and add the following code:

package main

import (
 "context"
 "fmt"
 "github.com/aws/aws-lambda-go/lambda"
)

// Define the input event structure if needed
type LambdaEvent struct {
 Name string `json:"name"`
}

// Handler function
func HandleRequest(ctx context.Context, event LambdaEvent) (string, error) {
 return fmt.Sprintf("Hello, %s!", event.Name), nil
}

func main() {
 lambda.Start(HandleRequest)
}

All AWS Lambda functions have some standardized arguments:

• Context: Metadata about the execution environment of the Lambda function
• Event: A predefined AWS event from a source such as S3, SQS, SNS, DynamoDB, API Gateway, Cloudwatch, AppSync, and so on

For reference, these are the inherent environment variables built in to each lambda function:

• AWS_LAMBDA_FUNCTION_NAMEThe function name
• AWS_LAMBDA_FUNCTION_VERSIONThe function version (e.g., $LATEST)
• AWS_LAMBDA_LOG_GROUP_NAMELog group for CloudWatch
• AWS_LAMBDA_LOG_STREAM_NAMELog stream for CloudWatch
• AWS_REGIONThe region where the function is running
• AWS_EXECUTION_ENVExecution environment (e.g., AWS_Lambda_go1.x)
• LAMBDA_TASK_ROOTThe root directory of the function code
• LAMBDA_RUNTIME_DIRDirectory where runtime files are stored

To create our process_file and process_row functions we’ll simply follow the steps we took for creating file_uploaded.

$ cd functions && \
  mkdir process_file && \
  mkdir process_row && \
  cd process_file && \
  go mod init process_file && \
  go get github.com/aws/aws-lambda-go/lambda && \
  touch main.go && \
  cd ../process_row && \
  go mod init process_row && \
  go get github.com/aws/aws-lambda-go/lambda && \
  touch main.go

Copy in the following main.go code for each function:

package main

import (
 "context"
 "fmt"
 "github.com/aws/aws-lambda-go/lambda"
)

// Define the input event structure if needed
type LambdaEvent struct {
 Name string `json:"name"`
}

// Handler function
func HandleRequest(ctx context.Context, event LambdaEvent) (string, error) {
 return fmt.Sprintf("Hello, %s!", event.Name), nil
}

func main() {
 lambda.Start(HandleRequest)
}

Run a quick cdk synth to sanity check everything. Provided there are no typos or errors, the cdk synth output should look like this:

Parameters:
  StageParam:
    Type: ...
     ...
  ServiceNameParam:
    Type: ...
    ...
Resources:
  S3Bucket00000000:
    Type: AWS::S3::Bucket

  DynamoDBTable00000XX0:
    Type: AWS::DynamoDB::Table

  BatchFilesDLQueue00XX000X:
    Type: AWS::SQS::Queue

  BatchFilesQueueXX00XXXX:
    Type: AWS::SQS::Queue
    
  BatchRowsDLQueueX00000X0:
    Type: AWS::SQS::Queue
    
  BatchRowsQueue00X0XXX0:
    Type: AWS::SQS::Queue
    
  FileUploadFunctionServiceRoleX00X00X0:
    Type: AWS::IAM::Role
    
  FileUploadFunction0X00X0X0:
    Type: AWS::Lambda::Function
    
  ProcessFileFunctionServiceRole00000000:
    Type: AWS::IAM::Role
   
  ProcessFileFunction00X00XX0:
    Type: AWS::Lambda::Function
    ...
    DependsOn:
      - ProcessFileFunctionServiceRole00000000
  ProcessRowFunctionServiceRole:
    Type: AWS::IAM::Role
    
  ProcessRowFunction0XXX00X0:
    Type: AWS::Lambda::Function
    Properties:
      Code:
        S3Bucket: cdk-xxx000xxx-assets-000000000000-us-east-1
        S3Key: 00000x00a999x99x0x000a99x99999xx0000000000aa9a9x9a909990900axa0x.zip
      Description: Process a CSV row
      Environment:
        Variables:
          BATCH_FILES_DLQ_NAME:
            Fn::Join:
              - ""
              - - Ref: ProjectParam
                - "-"
                - Ref: ServiceNameParam
                - "-"
                - Ref: StageParam
                - -batch-files-dl-queue
          BATCH_FILES_QUEUE_NAME:
            Fn::Join:
              - ""
              - - Ref: ProjectParam
                - "-"
                - Ref: ServiceNameParam
                - "-"
                - Ref: StageParam
                - -batch-files-queue
          BUCKET_NAME:
            Fn::Join:
              - ""
              - - Ref: ProjectParam
                - "-"
                - Ref: ServiceNameParam
                - "-"
                - Ref: StageParam
          DLQ_NAME:
            Fn::Join:
              - ""
              - - Ref: ProjectParam
                - "-"
                - Ref: ServiceNameParam
                - "-"
                - Ref: StageParam
                - -batch-rows-dl-queue
          QUEUE_NAME:
            Fn::Join:
              - ""
              - - Ref: ProjectParam
                - "-"
                - Ref: ServiceNameParam
                - "-"
                - Ref: StageParam
                - -batch-rows-queue
          TABLE_NAME: Users
      Handler: main
      Role:
        Fn::GetAtt:
          - ProcessRowFunctionServiceRole
          - Arn
    DependsOn:
      - ProcessRowFunctionServiceRole
    Metadata:
      aws:cdk:path: CsvMicroserviceStack/ProcessFileFunction/Resource
      aws:asset:path: asset.xxx000xxx0000000000xxx0x000a99x99999xx0000000000aa9a9x9a909990900axa0x
      aws:asset:is-bundled: false
      aws:asset:property: Code
  CDKMetadata:
    Type: AWS::CDK::Metadata
    Properties:
      Analytics: v2:deflate64:x/x+0+9/x+0/9+++x/0==
    Metadata:
      aws:cdk:path: CsvMicroserviceStack/CDKMetadata/Default

And our NewCsvMicroserviceStack should look like this:

func NewCsvMicroserviceStack(scope constructs.Construct, id string, props *CsvMicroserviceStackProps) awscdk.Stack {
 var sprops awscdk.StackProps
 if props != nil {
  sprops = props.StackProps
 }
 stack := awscdk.NewStack(scope, &id, &sprops)
 
 stageParameter := awscdk.NewCfnParameter(stack, jsii.String("StageParam"), &awscdk.CfnParameterProps{
  Type:        jsii.String("String"),
  Description: jsii.String("Is this a sandbox, gamma, beta, or prod environment?"),
  Default:     jsii.String("sandbox"),
 })
 
 serviceNameParameter := awscdk.NewCfnParameter(stack, jsii.String("ServiceNameParam"), &awscdk.CfnParameterProps{
  Type:        jsii.String("String"),
  Description: jsii.String("What do we want to call this microservice?"),
  Default:     jsii.String("microservice1"),
 })

 projectParameter := awscdk.NewCfnParameter(stack, jsii.String("ProjectParam"), &awscdk.CfnParameterProps{
  Type:        jsii.String("String"),
  Description: jsii.String("A unique name for your project"),
  Default:     jsii.String("deathramps"),
 })

 // Create the S3 bucket
 bucketName := fmt.Sprintf("%s-%s-%s", *projectParameter.ValueAsString(), *serviceNameParameter.ValueAsString(), *stageParameter.ValueAsString())
 awss3.NewBucket(stack, jsii.String("S3Bucket"), &awss3.BucketProps{
  BucketName:  jsii.String(bucketName),
 })

 // Create the DynamoDB Table
 tableName := "Users"
 awsdynamodb.NewTable(stack, jsii.String("DynamoDBTable"), &awsdynamodb.TableProps{
  TableName:    jsii.String(tableName),
  PartitionKey: &awsdynamodb.Attribute{
   Name:   jsii.String("id"),
   Type:   awsdynamodb.AttributeType_STRING,
  },
 });

 // Create the Batch Files SQS Queue
 batchFilesDLQueueName := fmt.Sprintf("%s-%s-%s-batch-files-dl-queue", *projectParameter.ValueAsString(), *serviceNameParameter.ValueAsString(), *stageParameter.ValueAsString())
 batchFilesDLQueue := awssqs.NewQueue(stack, jsii.String("BatchFilesDLQueue"), &awssqs.QueueProps{
  QueueName:       jsii.String(batchFilesDLQueueName),
  RetentionPeriod: awscdk.Duration_Days(jsii.Number(1)),
 })

 batchFilesQueueName := fmt.Sprintf("%s-%s-%s-batch-files-queue", *projectParameter.ValueAsString(), *serviceNameParameter.ValueAsString(), *stageParameter.ValueAsString())
 awssqs.NewQueue(stack, jsii.String("BatchFilesQueue"), &awssqs.QueueProps{
  QueueName:         jsii.String(batchFilesQueueName),
  VisibilityTimeout: awscdk.Duration_Seconds(jsii.Number(30)),
  DeadLetterQueue:   &awssqs.DeadLetterQueue{
            MaxReceiveCount: jsii.Number(3),
            Queue:           batchFilesDLQueue,
        },
 });

 // Create the Batch Rows SQS Queue
 batchRowsDLQueueName := fmt.Sprintf("%s-%s-%s-batch-rows-dl-queue", *projectParameter.ValueAsString(), *serviceNameParameter.ValueAsString(), *stageParameter.ValueAsString())
 batchRowsDLQueue := awssqs.NewQueue(stack, jsii.String("BatchRowsDLQueue"), &awssqs.QueueProps{
  QueueName:       jsii.String(batchRowsDLQueueName),
  RetentionPeriod: awscdk.Duration_Days(jsii.Number(1)),
 })

 batchRowsQueueName := fmt.Sprintf("%s-%s-%s-batch-rows-queue", *projectParameter.ValueAsString(), *serviceNameParameter.ValueAsString(), *stageParameter.ValueAsString())
 awssqs.NewQueue(stack, jsii.String("BatchRowsQueue"), &awssqs.QueueProps{
  QueueName:         jsii.String(batchRowsQueueName),
  VisibilityTimeout: awscdk.Duration_Seconds(jsii.Number(30)),
  DeadLetterQueue:   &awssqs.DeadLetterQueue{
            MaxReceiveCount: jsii.Number(3), // Messages will be moved to DLQ after 3 failed processing attempts
            Queue:           batchRowsDLQueue,
        },
 })

 // Create File Upload Lambda Function
 awslambda.NewFunction(stack, jsii.String("FileUploadFunction"), &awslambda.FunctionProps{
  Description:       jsii.String("Handle a file upload"),
  Runtime:           awslambda.Runtime_PROVIDED_AL2023(),
  Handler:           jsii.String("main"),
  Code:              awslambda.Code_FromAsset(jsii.String(filepath.Join("functions", "file_uploaded")), nil),
  Environment:       &map[string]*string{
   "BUCKET_NAME":   jsii.String(bucketName),
   "QUEUE_NAME":    jsii.String(batchFilesQueueName),
  },
 })

 // Create Process File Lambda Function
 awslambda.NewFunction(stack, jsii.String("ProcessFileFunction"), &awslambda.FunctionProps{
  Description: jsii.String("Process a CSV file"),
  Runtime:     awslambda.Runtime_PROVIDED_AL2023(),
  Handler:     jsii.String("main"),
  Code:        awslambda.Code_FromAsset(jsii.String(filepath.Join("functions", "process_file")), nil),
  Environment: &map[string]*string{
   "BUCKET_NAME": jsii.String(bucketName),
   "TABLE_NAME":  jsii.String(tableName),
   "QUEUE_URL": batchFilesQueue.QueueUrl(),
   "DL_QUEUE_URL": batchFilesDLQueue.QueueUrl(),
  },
 })

 return stack
}

You’ll notice that the YAML output has more resources that we didn’t explicitly create in the CDK code, namely the IAM roles for the Lambda functions. These will be helpful down the road when we need to expand the permissions of what the Lambda functions can do in order to process files, messages, and rows across different AWS resources. At this point we have specified all the main resources we are going to need and it’s time to once again run cdk deploy

Linking Resources and Granting Permissions

Now that we have both the S3 Bucket and the Lambda functions set up, we’ll also need to link them so that whenever there is a file with .csv extension uploaded to the S3 Bucket it will send a message to the Lambda File Uploaded function.

The first order of business is to import the libraries needed for this next part:

import (
 ...
"github.com/aws/aws-cdk-go/awscdk/v2/awss3notifications"
"github.com/aws/aws-cdk-go/awscdk/v2/awslambdaeventsources"
...

The second thing you will have to do is save the S3 Bucket and Lambda File Uploaded function resources to variables named bucket and fileUploadFunction. Then you can use those variables like so:

 bucket.GrantRead(fileUploadFunction, nil)
 bucket.AddEventNotification(
  awss3.EventType_OBJECT_CREATED,
  awss3notifications.NewLambdaDestination(fileUploadFunction),
  &awss3.NotificationKeyFilter{
   Suffix: jsii.String(".csv"),
  },
 )

The trigger event sent to the Lambda function will include metadata about the S3 event, including:

• Bucket name
• Key (file path) of the uploaded file
• Event time
• Size of the file

Next we want to set up the SQS link between the File Upload and Process File Lambda Functions:

batchFilesQueue.GrantSendMessages(fileUploadFunction)

bucket.GrantRead(processFileFunction, nil)

processFileFunction.AddEventSource(awslambdaeventsources.NewSqsEventSource(batchFilesQueue, &awslambdaeventsources.SqsEventSourceProps{
  BatchSize: jsii.Number(10),
 }))

batchRowsQueue.GrantSendMessages(processFileFunction)

batchFilesDLQueue.GrantPurge(processFileFunction)

Lastly we want to give the Process Rows Lambda function all the permissions it will need. Start by saving the DynamoDB table to a variable called table and then adding the following:

bucket.GrantRead(processRowFunction, nil)
table.GrantWriteData(processRowFunction)
batchRowsDLQueue.GrantPurge(processRowFunction)

Photo by Michael Dziedzic on Unsplash

Developing the AWS Lambda functions

At this stage we have created all the AWS cloud infrastructure that we’ll need; from here on out we mostly need to update the Lambda function source code. This will be the more interesting section.

File Uploaded

This function handles what we do when a file hits the file processing SQS queue. We want to open it and start reading off the rows. We create a read stream from the S3 object and set a batch limit of 25 items. Then we send these 25 items on to the next part of the app. We start with adding our imports and function signatures:

// file_uploaded
package main

import (
 "context"
 "encoding/csv"
 "fmt"
 "path/filepath"

 smithyhttp "github.com/aws/smithy-go/transport/http"
 "io"
 "log"
 "net/http"

 "github.com/aws/aws-lambda-go/events"
 "github.com/aws/aws-lambda-go/lambda"
 "github.com/aws/aws-sdk-go-v2/aws"
 "github.com/aws/aws-sdk-go-v2/config"
 "github.com/aws/aws-sdk-go-v2/service/s3"
)

// HandleRequest function
func HandleRequest(ctx context.Context, event events.S3Event) (map[string]interface{}, error) {
}

func main() {
  lambda.Start(HandleRequest)
}

Before adding any file handling logic we need to handle errors using the standard Go idiomatic pattern:

func HandleRequest(ctx context.Context, event events.S3Event) (map[string]interface{}, error) {
  response, err := doWork(ctx, event)
    if err != nil {
        log.Printf("Error: %v", err)
        return map[string]interface{}{
            "status": http.StatusBadRequest,
            "error":  err.Error(),
        }, nil
    }
    return response, nil
}

Next let’s extract the S3 bucket and key name (i.e. the file) from the event:

func doWork(ctx context.Context, event events.S3Event) (map[string]interface{}, error) {
    // Extract bucket and key from event
    fmt.Println(event)
    record := event.Records[0]
    fmt.Println(record)

    bucket := record.S3.Bucket.Name
    key := record.S3.Object.Key
    fmt.Println(bucket)
    fmt.Println(key)
    ...
}

Then we want to create an S3 client and fetch the file from the bucket:

  ...
  // Create AWS session
  cfg, err := config.LoadDefaultConfig(context.TODO())
  if err != nil {
   return nil, err
  }
  s3Client := s3.NewFromConfig(cfg, func(o *s3.Options) {
   o.APIOptions = append(o.APIOptions, smithyhttp.AddHeaderValue("X-Amz-Version", "2006-03-01"))
  })

  // Get object from S3
  params := &s3.GetObjectInput{
   Bucket: aws.String(bucket),
   Key:    aws.String(key),
  }

  result, err := s3Client.GetObject(context.TODO(), params)
  if err != nil {
   return nil, err
  }
  defer result.Body.Close()
  ...

Once we have the file, we want to set up a CSV file reader, read the header row, and process the rest of the CSV data in batches:

  ...
  // Read CSV data
  reader := csv.NewReader(result.Body)
  reader.Comma = ','
  reader.FieldsPerRecord = -1 // Allow variable number of fields

  // Read header
  headers, err := reader.Read()
  if err != nil {
   return nil, err
  }

  batchLimit := 25
  batch := []map[string]string{}

  for {
   record, err := reader.Read()
   if err == io.EOF {
    break
   }
   if err != nil {
    log.Printf("Error reading CSV: %v", err)
    return nil, err
   }

   // Convert record to map
   data := make(map[string]string)
   for i, value := range record {
    if i < len(headers) {
     data[headers[i]] = value
    }
   }

   batch = utils.DataHandler(data, batch, batchLimit, event, ctx)
  }
  ...

Here is where our first utility method comes into play. Once we have the file data as a map we want to validate the CSV row data is valid, then batch the records for DynamoDB and send them to the SQS queue for processing:

package utils

import (
 "context"
 "fmt"
 "os"
 "time"

 "github.com/aws/aws-lambda-go/events"
 "github.com/google/uuid"
)

import "file_uploaded/utils"

func DataHandler(data map[string]string, batch []map[string]string, batchLimit int, event events.S3Event, ctx context.Context) []map[string]string { // For now, we'll just append the data to the batch
 fmt.Println(data)

 isValid, errorMsg := validateRow(data)

 if isValid {
  // Generate UUID v1
  id, _ := uuid.NewUUID()

  // Get TTL value
  ttlValue := getTTL()

  // Create new record with ID, TTL and original data
  record := map[string]string{
   "id":  id.String(),
   "ttl": ttlValue,
  }

  // Copy all fields from data to record
  for key, value := range data {
   record[key] = value
  }

  batch = append(batch, record)

  if len(batch) == batchLimit {
   processBatch(batch, event, ctx)
   batch = []map[string]string{}
  }
 } else {
  // Send error row to DLQ
  SendToQueue(errorMsg, ctx, os.Getenv("DL_QUEUE_URL"))
 }

 return batch
}

The utility functions called by DataHandler include ProcessBatch :

package utils

import (
 "context"
 "encoding/json"
 "fmt"
 "os"

 "github.com/aws/aws-lambda-go/events"
)

// ProcessBatch processes a batch of records by sending them to an SQS queue
func ProcessBatch(batch []map[string]string, event events.S3Event, ctx context.Context) {
 // Log the batch being processed
 batchJSON, _ := json.Marshal(batch)
 fmt.Printf("Processing batch: %s\n", string(batchJSON))
 
 // Create batch items in the format expected for DynamoDB
 batchItems := make([]map[string]map[string]map[string]string, len(batch))
 for i, record := range batch {
  batchItems[i] = map[string]map[string]map[string]string{
   "PutRequest": {
    "Item": record,
   },
  }
 }
 
 // Get table name from environment variable
 tableName := os.Getenv("TABLE_NAME")
 
 // Create message body
 messageBody := map[string]interface{}{
  "tableName":  tableName,
  "batchItems": batchItems,
 }
 
 // Send to queue
 SendToQueue(messageBody, ctx, os.Getenv("QUEUE_URL"))
}

Lastly, we can process any remaining records in the final batch via utils.EndHandler and return a response:

utils.EndHandler(batch, event, ctx)

  return map[string]interface{}{
   "status": http.StatusOK,
   "body":   "process finished",
  }, nil

Row Processing

This function handles what we do when a batch of rows hits the row processing SQS queue. We want to validate its content and attempt to write it to DynamoDB. An important feature of this function is it implements a retry mechanism with exponential backoff for any unprocessed items. If any items fail to write, the function will automatically attempt to re-write them with increasing delay intervals.

package main

import (
 "context"
 "encoding/json"
 "fmt"
 "log"
 "math"
 "os"
 "time"

 "github.com/aws/aws-lambda-go/events"
 "github.com/aws/aws-lambda-go/lambda"
 "github.com/aws/aws-sdk-go-v2/aws"
 "github.com/aws/aws-sdk-go-v2/config"
 "github.com/aws/aws-sdk-go-v2/service/dynamodb"
 "github.com/aws/aws-sdk-go-v2/service/dynamodb/types"
)

const (
 maxRetries      = 3
 baseBackoffMs   = 100
 maxBatchItems   = 25 // DynamoDB BatchWriteItem limit
)

type BatchMessage struct {
 TableName  string                                    `json:"tableName"`
 BatchItems []map[string]map[string]map[string]string `json:"batchItems"`
}

// HandleRequest processes SQS messages containing batches of DynamoDB write requests
func HandleRequest(ctx context.Context, sqsEvent events.SQSEvent) error {
 cfg, err := config.LoadDefaultConfig(ctx)
 if err != nil {
  return fmt.Errorf("failed to load AWS config: %w", err)
 }

 dynamoClient := dynamodb.NewFromConfig(cfg)
 
 for _, record := range sqsEvent.Records {
  if err := processRecord(ctx, dynamoClient, record); err != nil {
   // Log but don't fail the entire batch - let SQS handle retries
   log.Printf("Failed to process record %s: %v", record.MessageId, err)
   return err // This will cause SQS to retry or send to DLQ
  }
 }

 return nil
}

func processRecord(ctx context.Context, client *dynamodb.Client, record events.SQSMessage) error {
 var msg BatchMessage
 if err := json.Unmarshal([]byte(record.Body), &msg); err != nil {
  return fmt.Errorf("invalid message format: %w", err)
 }

 // Validate batch size
 if len(msg.BatchItems) > maxBatchItems {
  return fmt.Errorf("batch size %d exceeds DynamoDB limit of %d", 
   len(msg.BatchItems), maxBatchItems)
 }

 // Convert to DynamoDB format
 writeRequests := make([]types.WriteRequest, 0, len(msg.BatchItems))
 for _, item := range msg.BatchItems {
  putRequest, ok := item["PutRequest"]
  if !ok {
   continue
  }
  
  itemData, ok := putRequest["Item"]
  if !ok {
   continue
  }

  // Convert map[string]string to map[string]types.AttributeValue
  avMap := make(map[string]types.AttributeValue)
  for k, v := range itemData {
   avMap[k] = &types.AttributeValueMemberS{Value: v}
  }

  writeRequests = append(writeRequests, types.WriteRequest{
   PutRequest: &types.PutRequest{
    Item: avMap,
   },
  })
 }

 // Write with exponential backoff retry for unprocessed items
 return writeWithRetry(ctx, client, msg.TableName, writeRequests)
}

func writeWithRetry(ctx context.Context, client *dynamodb.Client, tableName string, 
 requests []types.WriteRequest) error {
 
 unprocessed := requests
 attempt := 0

 for len(unprocessed) > 0 && attempt < maxRetries {
  if attempt > 0 {
   // Exponential backoff with jitter
   backoff := time.Duration(baseBackoffMs*int(math.Pow(2, float64(attempt)))) * time.Millisecond
   jitter := time.Duration(float64(backoff) * 0.1) // 10% jitter
   time.Sleep(backoff + jitter)
   
   log.Printf("Retry attempt %d for %d unprocessed items", attempt, len(unprocessed))
  }

  input := &dynamodb.BatchWriteItemInput{
   RequestItems: map[string][]types.WriteRequest{
    tableName: unprocessed,
   },
  }

  result, err := client.BatchWriteItem(ctx, input)
  if err != nil {
   return fmt.Errorf("batch write failed on attempt %d: %w", attempt, err)
  }

  // Check for unprocessed items (throttling)
  if result.UnprocessedItems != nil {
   if items, ok := result.UnprocessedItems[tableName]; ok {
    unprocessed = items
    log.Printf("DynamoDB throttled: %d items unprocessed", len(unprocessed))
   } else {
    unprocessed = nil
   }
  } else {
   unprocessed = nil
  }

  attempt++
 }

 if len(unprocessed) > 0 {
  return fmt.Errorf("failed to write %d items after %d retries - likely capacity exceeded", 
   len(unprocessed), maxRetries)
 }

 return nil
}

func main() {
 lambda.Start(HandleRequest)
}

The critical detail here is handling DynamoDB’s unprocessed items. When you exceed provisioned throughput, DynamoDB doesn’t throw an error — it returns a 200 response with theUnprocessedItems field populated. Missing this can cost you hours of debugging so we make sure to check it here.

The exponential backoff prevents cascading failures. Without it, retries hammer DynamoDB at the same rate that caused the throttling in the first place. The 10% jitter prevents synchronized retries across concurrent Lambda invocations. Items that couldn’t be written immediately are retried with increasing delays, allowing the system to self-regulate without dropping data.

Utilizing the Dynamo table for data storage and management

DynamoDB is a NoSQL database which instantly makes you sound smarter and is a immediate cheat code to success (not really). Instead of primary keys like in relational database management services, Dynamo relies on partition keys and optionally a sort key to create your analogous primary key. You want to set your table up this way in order to be able to query it efficiently. DynamoDB billing can come back to bite you if you don’t set it up right from the beginning, so it’s worth investing a little time here.

For the purposes of this tutorial I’ve set us up with only 1 provisioned read/write capacity units. If you want to query the records in your User table there are many options but the 2 easiest ones are through the AWS web console or the AWS CLI.

Photo by Joan Gamell on Unsplash

Deploying and Testing the Microservice

Before deploying to AWS, let’s ensure everything is properly configured. Run these validation commands from your project root

# Validate your CDK code
cdk synth

# Check for any security issues
cdk doctor

# Review what will be deployed
cdk diff

If cdk synth still completes without errors and generates the CloudFormation template, we’re good to deploy.

Lambda functions written in Go need to be compiled for the Linux environment that AWS Lambda uses. Create a build script to automate this process. In your project root, create a file called build.sh:

#!/bin/bash

# Build all Lambda functions
for dir in functions/*/; do
    echo "Building $(basename $dir)..."
    cd "$dir"
    GOOS=linux GOARCH=amd64 go build -o bootstrap .
    cd ../..
done

echo "All functions built successfully"

Make it executable and run it:

chmod +x build.sh
./build.sh

This creates a bootstrap executable in each function directory, which is what Lambda expects for custom runtimes.

Now we’re ready to deploy. The first deployment will take several minutes as CDK provisions all resources:

# Deploy with parameter overrides
cdk deploy \
  --parameters ProjectParam=myproject \
  --parameters ServiceNameParam=csvimport \
  --parameters StageParam=sandbox

During deployment, you’ll see output showing each resource being created. CDK will ask for confirmation before making changes — review the security IAM statement changes carefully before proceeding. Once complete, you’ll see output similar to:

✅  CsvMicroserviceStack
Stack ARN:
arn:aws:cloudformation:us-east-1:123456789012:stack/CsvMicroserviceStack/...
Outputs:
CsvMicroserviceStack.BucketName = myproject-csvimport-sandbox
CsvMicroserviceStack.TableName = Users

Save these output values, you’ll need them for testing.

Setting Up Test Scenarios

I’ve included a test CSV file with valid user data in the repo called users.csv .The simplest way to test is to upload this file directly to S3:

# Upload the test file
aws s3 cp test-users.csv s3://myproject-csvimport-sandbox/test-users.csv

# Watch the logs in real-time
aws logs tail /aws/lambda/CsvMicroserviceStack-FileUploadFunction --follow

You should see log entries showing:

1. File upload detected
2. CSV parsing initiated
3. Batches sent to SQS
4. Rows processed
5. DynamoDB writes completed

Verifying Data in DynamoDB

Once processing completes, verify the data landed correctly:

# Scan the table to see all records
aws dynamodb scan --table-name Users --max-items 10

# Query a specific record
aws dynamodb get-item \
  --table-name Users \
  --key '{"id": {"S": "UUID"}}'

Or use the AWS Console: Navigate to DynamoDB → Tables → Users → Explore table items.

Best Practices and Optimization

As tedious as this was, it’s by no means a production-ready application. Here are some best practices and optimizations that can be introduced to make it more robust:

• Configure monitoring and logging using AWS CloudWatch
• Set up alerts for critical events in CloudWatch using performance thresholds
• Implement Least Privilege IAM Policies
• Enable Encryption for all resources: S3 Bucket, DynamoDB Table, SQS Queue
• Restrict S3 Bucket Access
• Add VPC Configuration (for sensitive workloads)
• Implement S3 Lifecycle Policies to automatically delete or archive processed files
• Optimize Lambda Memory and Timeout; Higher memory = faster execution = lower cost (usually). You can even use AWS Lambda Power Tuning to find optimal configuration automatically
• Optimize Batch Sizes; Larger batches = fewer Lambda invocations = lower cost, but also longer processing times
• Use Lambda Reserved Concurrency Strategically; prevent one function from consuming all account concurrency
• Implement CSV Streaming for files larger than Lambda’s memory limit (up to 10GB), stream processing
• Optimize Go Lambda Cold Starts

Cost Estimation

For an example production workload processing 1 million rows per day:

Cost Table

Conclusion

Large messaging queues and async processing are common features to tackle in any architecture. What makes this one unique is the source of data; we can always tie it back to the exact file and row it came from. If you enjoyed this exploratory project of AWS and Golang I would encourage you to build some more customizations on top this microservice and see where it takes you. This tutorial was good for getting started, but would need more work to be truly usable for production deployment. To check out the complete code please visit my github profile:

https://github.com/beatleboy501/CSVMicroservice

If you found this article insightful be sure to smash that 👏 button and follow 👉 https://medium.com/@andrewrobertallison