Building a Scalable Serverless Architecture with AWS: A Step-by-Step Guide for Mini Project
Introduction
In the realm of serverless architecture, building scalable and efficient systems is key to success. Join us on a journey as I explore the implementation of “Sync Service,” a mini project designed using AWS services. In this step-by-step guide, I’ll leverage the power of AWS services, including API Gateway, DynamoDB, Lambda, SQS, and OpenSearch, to create a robust system for data synchronization.
Section 1: Project Overview
“Sync Service” focuses on the effective management and search of book metadata through a series of AWS services. I’ll utilize API Gateway for handling incoming requests, DynamoDB as our NoSQL database, Lambda functions for asynchronous data processing, SQS for messaging, and OpenSearch for scalable and powerful search capabilities. Below is a high-level architectural diagram depicting the key components and their interactions
Section 2: Setting Up the Serverless Framework
To kick things off, let’s set up our project using the Serverless Framework. This powerful tool simplifies the deployment and management of serverless applications.
Now, if you’re unfamiliar with serverless or need a deeper dive into its capabilities, I recommend exploring the Serverless Framework Documentation for a comprehensive guide. This will provide you with the foundational knowledge needed to make the most out of the “Sync Service” project.
Section 3: Implementing the API Gateway and DynamoDB Integration
Our journey continues with the creation of an API Gateway endpoint for adding products, and DynamoDB will serve as our robust data store. Dive into the details of this implementation in the serverless.yml file and corresponding Lambda Function.
# yml file
functions:
createPorduct:
handler: src/handler.createProduct
role: ${env:IAM_ROLE}
events:
- http:
method: post
path: force-sync// create lambda function
module.exports.createProduct = async (event) => {
let data = JSON.parse(event.body);
try {
const createdId = await DynamoDb.put(data)
data.id = createdId;
return statusCode(201, data);
} catch (error) {
return statusCode(500, error.message);
}
}const { DynamoDBClient } = require("@aws-sdk/client-dynamodb");
const { PutCommand, DynamoDBDocumentClient } = require("@aws-sdk/lib-dynamodb");
const uuid = require('uuid');
const DBClient = new DynamoDBClient({});
const docClient = DynamoDBDocumentClient.from(DBClient);
const put = async (data) => {
const itemId = uuid.v4();
try {
let params = {
TableName: process.env.TABLE_NAME,
Item: {
productId: itemId,
modelType: data.modelType,
state: data.state,
name: data.name,
doi: data.doi,
productCode: data.productCode,
edition: data.edition,
publicationType: data.publicationType
},
ConditionExpression: "attribute_not_exists(productId)",
}
const command = new PutCommand(params);
await docClient.send(command);
return itemId;
} catch (error) {
throw error
}
}
module.exports = {
put
}Section 4: DynamoDB Stream and SQS Integration
Discover the power of DynamoDB Streams to capture changes in our database. Utilize a Lambda function to listen to these changes and send updates to an SQS queue. Explore the setup in the serverless.yml file and Lambda Function to achieve seamless integration.
# yml file for dynamo DB
productTable:
Type: AWS::DynamoDB::Table
Properties:
TableName: ${env:TABLE_NAME}
BillingMode: PAY_PER_REQUEST
AttributeDefinitions:
- AttributeName: productId
AttributeType: S
KeySchema:
- AttributeName: productId
KeyType: HASH
StreamSpecification:
StreamViewType: NEW_IMAGE
#yml for lambda
triggerDynamoDB:
handler: src/handler.getDynamoTableData
role: ${env:IAM_ROLE}
events:
- stream:
type: dynamodb
arn: !GetAtt [productTable, StreamArn]//lambda function
module.exports.getDynamoTableData = async (event) => {
const product = event.Records[0].dynamodb
const productData = convertData(product.NewImage)
try {
const response = await SQS.pushToSQS(productData);
console.log(JSON.stringify(response))
return statusCode(200, response)
} catch (error) {
console.log(JSON.stringify(error.message))
return statusCode(500, error.message)
}
}const { SQSClient, SendMessageCommand, DeleteMessageCommand } = require("@aws-sdk/client-sqs");
const sqs = new SQSClient({ region: "us-east-1" });
const queueUrl = process.env.QUEUE_URL;
const pushToSQS = async (data) => {
try {
const params = {
MessageBody: JSON.stringify(data),
QueueUrl: queueUrl,
};
const command = new SendMessageCommand(params)
console.log("--command")
console.log(JSON.stringify(command))
return await sqs.send(command);
} catch (error) {
throw error
}
}
module.exports = {
pushToSQS
}Section 5: Processing SQS Messages with Lambda
The journey continues with a second Lambda function triggered by SQS messages. This function will process messages and store data in OpenSearch.
# yml file for SQS
MyQueue:
Type: AWS::SQS::Queue
Properties:
QueueName: product-topic
VisibilityTimeout: 30
MaximumMessageSize: 2048
MessageRetentionPeriod: 345600 #yml for lambda function
saveDataFromSQSToElastic:
handler: src/handler.saveToElastic
role: ${env:IAM_ROLE}
events:
- sqs:
arn: !GetAtt MyQueue.Arn//lambda function
module.exports.saveToElastic = async (event) => {
console.log(event.Records[0].body)
if (event.Records && event.Records[0] && event.Records[0].body) {
const receiptHandle = event.Records[0].receiptHandle;
const productData = JSON.parse(event.Records[0].body)
console.log(productData)
if (validator.validateModelType(productData)) {
console.log("Invalid Model Type")
return statusCode(400, "Invalid Model Type")
}
try {
const response = await opensearch.indexCreation(productData);
console.log(JSON.stringify(response))
console.log("index creation is done")
} catch (error) {
console.log(error)
}
try {
const delResponste = await SQS.deleteMessage(receiptHandle)
console.log(delResponste)
console.log("deleting message is done")
} catch (error) {
console.log(error)
}
}
else {
return statusCode(500, "Invalid Data")
}
}Section 6: Section 6: Utilizing OpenSearch for Indexing
I am utilizing the OpenSearch URL created through the AWS console. Explore the Lambda function designed to accomplish OpenSearch indexing.
const { Client } = require('@opensearch-project/opensearch');
require('dotenv').config();
const INDEX_NAME = process.env.INDEX_NAME;
const openSearchUsername = process.env.OPENSEARCH_USERNAME;
const openSearchPassword = process.env.OPENSEARCH_PASSWORD;
const openSearchEndpoint = process.env.OPENSEARCH_ENDPOINT;
const client = new Client({
node: openSearchEndpoint,
auth: {
username: openSearchUsername,
password: openSearchPassword
}
});
const indexCreation = async (product) => {
try {
await client.index({
index: INDEX_NAME,
id: product.id,
body: product
})
return {
statusCode: 200,
message: 'Document ' + product.id + ' indexed successfully.'
};
} catch (error) {
console.error('Error indexing document: ', error);
return {
statusCode: 500,
message: 'Error indexing document: ' + error.message
};
}
}
module.exports = {
indexCreation
}Section 7: Testing and Monitoring
Effective testing and monitoring are crucial for the success of any serverless application. Utilize tools like AWS SAM for local testing and AWS CloudWatch for monitoring Lambda functions.
This architecture ensures a seamless flow of book metadata, allowing for scalability, fault tolerance, and efficient data processing. Notably, the integration of OpenSearch plays a crucial role in enhancing the search capabilities of “Sync Service,” enabling effective and fast retrieval of book metadata stored in DynamoDB through indexing.
If you have any questions or doubts regarding the code implementation, feel free to explore the GitHub repository associated with this project.
Thank You and Happy Coding !
