Large-scale Data Processing with Step Functions : AWS Project
Introduction
In this project, we tackle the challenge of orchestrating large-scale data processing using AWS Step Functions. By integrating key AWS services like Amazon S3, IAM, CloudWatch, and AWS X-Ray, we build a scalable, secure, and optimized workflow for handling vast amounts of data. This setup is designed to reduce operational complexity, enhance scalability, and provide robust monitoring, making it ideal for data-intensive industries requiring high levels of efficiency and control.
Tech Stack
Key AWS services, tools, and technologies used in this project include:
AWS Step Functions: Orchestrates the data processing workflow
Amazon S3: Provides centralized storage for input and output data
IAM (Identity and Access Management): Ensures secure permissions and data access policies
CloudWatch: Monitors each step of the workflow, logging actions and errors
AWS X-Ray: Provides detailed tracing and performance insights into the workflow’s execution
Prerequisites
To follow along with this project, you’ll need the following prerequisites:
AWS Knowledge: Familiarity with AWS services like S3, IAM, and CloudWatch.
Serverless and Workflow Concepts: Basic understanding of serverless architecture and workflow orchestration.
AWS CLI and SDKs: Installed and configured for managing AWS resources.
IAM Permissions: Ensure the necessary IAM roles and policies for accessing and executing Step Functions.
Problem Statement or Use Case
Organizations often deal with complex workflows that require efficient processing of large data sets. This project addresses the need for a scalable and automated solution to manage such workflows reliably. Key challenges include:
Coordination of Multiple Services: Processing large datasets often involves multiple tasks and services that need to be coordinated systematically.
Scalability: As data size grows, the system must scale to handle the increased load.
Visibility and Monitoring: Detailed logging, tracing, and monitoring are essential for troubleshooting and optimizing the workflow.
AWS Step Functions, combined with other AWS services, provides an ideal solution for orchestrating complex workflows in a serverless environment. This architecture not only improves processing efficiency but also enhances visibility and simplifies troubleshooting.
Architecture Diagram
Below is the architecture diagram for the data processing workflow. The visual aid below illustrates how AWS Step Functions orchestrates interactions among S3, IAM, CloudWatch, and AWS X-Ray to create a reliable and traceable workflow.
Important
Follow the instructions on this page only if you are executing this workshop in your own account. To skip these instructions
Europe (Ireland) eu-west-1
Asia Pacific (Sydney) ap-southeast-2
- On the Specify stack details page, Stack name will be auto-populated to sfw-hello-distributed-map. (You can enter a different name if you want.) Click Next two times.
- Wait until the stack shows CREATE_COMPLETE status.
Important
Follow the instructions on this page only if you are executing this workshop in your own account. To skip these instructions
Europe (Ireland) eu-west-1
Asia Pacific (Sydney) ap-southeast-2
- On the Specify stack details page, Stack name will be auto-populated to sfw-processmulti-distributed-map. (You can enter a different name if you want.) Click Next two times.
- Wait until the stack shows CREATE_COMPLETE status.
Important
Follow the instructions on this page only if you are executing this workshop in your own account. To skip these instructions
Europe (Ireland) eu-west-1
Asia Pacific (Sydney) ap-southeast-2
- On the Specify stack details page, Stack name would be auto populated to sfw-optimization-distributed-map. You can specify a different name if you want.
- Wait till the stack shows CREATE_COMPLETE status.
Important
Follow the instructions on this page only if you are executing this workshop in your own account. To skip these instructions
Europe (Ireland) eu-west-1
Asia Pacific (Sydney) ap-southeast-2
- On the Specify stack details page, Stack name would be auto populated to sfw-healthcare-processing. You can specify a different name if you want.
- Wait till the stack shows CREATE_COMPLETE status.
Important
Follow the instructions on this page only if you are executing this workshop in your own account. To skip these instructions
Europe (Ireland) eu-west-1
Asia Pacific (Sydney) ap-southeast-2
Important
Follow the instructions on this page only if you are executing this workshop in your own account. To skip these instructions
The solution is deployed in two sets. The first stack will deploy a stack of resources that will generate our simulated dataset. It is orchestrated by Step Functions and consists of three Lambda Functions that will generate the data as well as generate a simulated S3 Inventory Report. The second stack will execute the first state machine as well as deploy the components for the module.
Europe (Ireland) eu-west-1Asia Pacific (Sydney) ap-southeast-2
- Click on the Launch link against any of the regions in the table below to start the deployment.
Region Deployment
US East (N. Virginia) us-east-1
Asia Pacific (Singapore) ap-southeast-1
Take Defaults — Click Next 3 times and Submit
Deployment Screenshots
Module 1 — Basics
Introduction to Distributed Map
from
Module 1 — Basics
Introduction to Distributed Map
from
Module 1 — Basics
Introduction to Distributed Map
from
from
The workflow iterates on the electronics review data and filters highly rated reviews. You will download the data to S3, run the workflow, and verify the results. In the process, you will learn how to build and run a simple distributed map workflow yourself.
— Serverless visual workflow service
Reviewing the Workflow
Open the
Take a closer look at the map definition. You define it as DISTRIBUTED to tell Step Functions to run the map state in distributed mode.
Read ahead and you will notice that there are a few more settings.
Firstly, you can do batching. Do you see MaxItemsPerBatch set as 1000?
You can not only run 10,000 (10K) workflows, you can also batch the data to each workflow which means, in a single iteration, you can process 10K * 1K = 10M records from the csv file!
Secondly, you can write the output of the distributed map or the child workflow execution results to an S3 location in an aggregated fashion.
Thirdly, you can set failure toleration. What does that mean? You don’t want to run 100M records when half of them are bad data. It is a waste of time and money to process those records. By default, the failure toleration is set to 0. Any single child workflow failure will result in the failure of the workflow.
Data quality is a big challenge with large data processing. So, you can set a percentage or number of items that can be tolerated as failures. When failures exceed that tolerance, the Step Functions workflow fails, saving you time and money.
The states inside the distributed map are run as separate child workflows. The number of child workflows is dependent on the concurrency setting and the volume of the records to process. For example, you might set the concurrency to 1000 and batch size to 100, but if the total number of records in the file is just 20K, Step Functions only needs 200 child workflows (20,000 / 100 = 200). On the other hand, if the file has 200K records, Step Functions will spin up 1000 child workflows to reach the max concurrency and as child workflows complete, Step Functions will spin up child workflows until all 2000 (200,000 /100 = 2000) child workflows are completed.
Running the Workflow
--output df_electronics.csv
Windows (PowerShell)
Invoke-WebRequest then select the bucket containing “hellodmapdatabucket” in its name.
Copy the downloaded file “df_electronics.csv” to the S3 bucket.
, select and open the state machine containing HelloDmapStateMachine in its name.
Choose Start execution in the top right corner.
In the popup, enter the following input:
{
"key":"df_electronics.csv",
"output":"results"
}
You are providing the name of the file and S3 prefix where you want the results of the distributed map to be stored.
Click Start execution.
In a few seconds, you will see the execution start to run. It takes a couple of minutes to complete the processing.
Navigate to the bottom of the workflow execution page and click Map Run.
Open one of the child workflows and view the execution input/output. You see in the output window that the pass state filtered the records with ratings of 4 and above.
Recall you also stored the results of the workflow to an S3 bucket! Open the
Navigate to contents inside results prefix, select "SUCCEEDED_0.json", and download the file to view the results. You notice that the content of the file is the aggregated result of all the child workflows. If you are building map reduce use cases, this content can be used for the downstream. To learn more about how result writer works, follow the link and
In the previous module, you saw an example of distributed processing with a single S3 object. Distributed map not only iterates on a single large object located in S3, it can also iterate on a collection of objects in S3. You can iterate through and process each object in parallel and aggregate the results. This supports various use cases, such as processing thousands of log files, a Monte Carlo simulation which runs the same processing for multiple inputs, running a backfill process that scans millions of files for security vulnerability for past dates.
Below diagram shows how distributed map works for multiple S3 objects. Notice that it uses S3.listObjectV2 instead of S3.GetObject which is used in the previous sub module. When processing multiple objects, distributed map lists the metadata of the objects, distributes batches of the metadata to the child workflows. This means you can process any file format; structured, unstructured, and semi-structured.
In this module, you are going to build a distributed map workflow that processes thousands of weather data files from
— Compute service; functions in serverless runtimes
To quickly build the workflow, we have created a few resources ahead of time.
Lambda function to find the highest precipitation for the station
One S3 bucket for the dataset and another S3 bucket for storing distributed map results
Sample dataset of 1,000 S3 objects from NOAA climatology data
Amazon DynamoDB table to store the precipitation data
Building the Workflow
then choose Create state machine button.
Choose the Blank card and choose Select.
You are in Workflow Studio. Take a moment to explore it. You will see the actions, flows, and patterns on the left side. You can drag the states on the left to the center of the page where you see the workflow design. You can configure the input, output, errors, etc. on the right side of the UI. If you click the Definition toggle, you can view the ASL definition of the workflow.
Take some time to explore the menus.
Alright. You are now going to configure additional attributes of the distributed map. First, you will configure where to read the dataset from. You will pass the location of the precreated dataset in S3 as input.
Select Amazon S3 as the Item source.
Select S3 object list in S3 item source dropdown.
Select Get bucket and prefix at runtime from state input in S3 bucket dropdown.
For Bucket Name, enter $.bucket
For Prefix, enter $.prefix
Leave everything else as default and move on to adding the child workflow components.
Enter Lambda
in the search textbox at the top left.
Drag and drop the Lambda — Invoke action to the center.
Click on the Function name dropdown and select function with HighPrecipitation in its name.
then copy the full name of the bucket containing MultiFileDataBucket in its name.
Return to the Start execution popup and enter the following json as input, replacing the bucket name with your bucket name from S3:
CODE{
"bucket": "bucketname",
"prefix":"csv/by_station"
}
To verify the ASL definition
CODE{
"Comment": "A description of my state machine",
"StartAt": "Map",
"States": {
"Map": {
"Type": "Map",
"ItemProcessor": {
"ProcessorConfig": {
"Mode": "DISTRIBUTED",
"ExecutionType": "STANDARD"
},
"StartAt": "Lambda Invoke",
"States": {
"Lambda Invoke": {
"Type": "Task",
"Resource": "arn:aws:states:::lambda:invoke",
"OutputPath": "$.Payload",
"Parameters": {
"Payload.$": "$",
"FunctionName": "arn:aws:lambda:{region}:{account}:function:sfw-processmulti-distribu-HighPrecipitationFunctio-vH7XVagF8llI:$LATEST"
},
"Retry": [
{
"ErrorEquals": [
"Lambda.ServiceException",
"Lambda.AWSLambdaException",
"Lambda.SdkClientException",
"Lambda.TooManyRequestsException"
],
"IntervalSeconds": 1,
"MaxAttempts": 3,
"BackoffRate": 2
}
],
"End": true
}
}
},
"End": true,
"Label": "Map",
"MaxConcurrency": 1000,
"ItemReader": {
"Resource": "arn:aws:states:::s3:listObjectsV2",
"Parameters": {
"Bucket.$": "$.bucket",
"Prefix.$": "$.prefix"
}
},
"ItemBatcher": {
"MaxItemsPerBatch": 100,
"BatchInput": {
"Bucket.$": "$.bucket"
}
}
}
}
}
By default, distributed map fails even when a single child fails. Let’s explore what actually caused the child workflow to fail.
Click Map Run to check the child workflow executions.
It looks like Lambda is expecting event[BatchInput][Bucket] and it is not found. Explore the input to the Lambda function by selecting Execution input and output at the top.
Save the workflow and choose Execute button.
Do not forget to execute the workflow with the proper bucket and prefix input. If you would like, you can get the input by navigating to the previous execution of the workflow.
Voila!! It is a success now!
Viewing the Workflow Results
and select the function containing HighPrecipitation in its name.
Explore the Code.
The Lambda function writes the calculated value to DynamoDB table.
CODEdef _write_results_to_ddb(high_by_station: Dict[str, Dict]):
dynamodb = boto3.resource("dynamodb")
table = dynamodb.Table(os.environ["RESULTS_DYNAMODB_TABLE_NAME"])
The table name comes from the env variable RESULTS_DYNAMODB_TABLE_NAME.
Click Configuration and select Environment Variable to view the table name.
then select Tables from left side menu.
Select the table name that you saw in the Lambda configuration and choose Explore table items. You can now view the calculated highest precipitation across stations.
. With S3 listObjectsV2, Step Functions is making S3 listObjectsV2 *API calls on your behalf to retrieve all of the items needed to run the Distributed Map. Each call to *listObjectsV2 can only return a maximum of 1000 S3 objects. This means that if you have 2,000,000 objects to process, Step Functions has to make at least 2000 API calls. This API is fast and it won’t take too long, but if you have an S3 Inventory file that has all the objects listed in it that you need to process, you can use that as the input.
Using an S3 Inventory file as the input for a Distributed Map when processing large numbers of files is faster than S3 listObjectsV2. This is because, for S3 Inventory ItemReaders, there is a single S3 getObject call to get the manifest file and then one call for each Inventory file. If you know that your Distributed Map is going to run on a set schedule you can schedule the S3 Inventory to be created ahead of time.
Module 2 — Advanced
Welcome to the Advanced module of the data processing workshop!
In the
In this module, we use the same example workflow from earlier sub module
— compute service; functions in serverless runtimes
A Lambda function to find the highest precipitation for the station.
One S3 bucket for data set and another S3 bucket for storing distributed map results.
Sample data set of 1000 S3 objects from
Using the precreated Step Functions workflow, you will tune some attributes/fields of distributed map and understand the performance and cost impact of the change.
Choosing the Workflow Type
.
When you use distributed map, Step Functions spins up child workflows to run the states inside the distributed map. The number of child workflows to spin up is dependent on the number of objects or records to process, batch size and concurrency. You can define the child workflow to run as either standard or express based on your use case.
In the following sections, you will learn how to change workflow type of distributed map child workflows, try out a technique to find if express workflow suits your use case, and the cost impact of running standard vs express.
, select State machines from the right menu.
Select the workflow that starts with OptimizationStateMachine.
Choose edit button to edit the workflow in workflow studio.
Review the definition in the workflow studio by enabling definition at the right.
Highlight the Distributed map high precipitation step in the workflow graphic.
Child workflow can be run either STANDARD or EXPRESS. Express workflows are generally less expensive and run faster than Standard workflows.
Sometimes, you may not be sure if your workflow runs within 5 minutes. In this section, you are going to use a feature of distributed map that allows you to test your data with small number of items. This technique is helpful in couple of ways
To determine the duration of the child workflow
To gain confidence that the child workflow logic will run fine when running with full data set.
Start making the changes
Toggle Definition button to edit the configuration.
Expand Additional configuration and select Limit number of items.
Type 1 in Max Items textbox.
- Observe the duration for single item. It is around 3 seconds.
Return to workflow studio.
Change the workflow type to EXPRESS.
Consider you are processing 500K objects and set the batch size to 500.
500k objects / 500 objects per workflow = 1000 child workflows
Distributed map runs a total of 1000 child workflows to process 500K objects.
With Express workflows, you pay for the number of requests for your workflow and the duration. With scenario outlined earlier under Review cost impact, we need additional dimension of how long workflow runs to calculate the express workflow cost. Let's assume express child workflow runs for an average of 100sec to process 500 objects using 64-MB memory.
Duration cost = (Avg billed duration ms / 100) * 0.0000001042
Duration cost = (100,000 MS /100) * $ 0.0000001042 = $0.0001042
Express request cost = $0.000001 per request ($1.00 per 1M requests)
workflow cost = (Express request cost + Duration cost) x Number of Requests
workflow cost = ($0.000001 + $0.0001042) x 1000 = $0.10
Express child workflow introduces 1 state transition per child workflow regardless of how many states you have inside the workflow. This is to start each child execution.
Transition cost = (1 * 1000) x $0.000025 = $0.025
Total cost = $0.10 + $0.025 = $0.125
If we repeat the calculation for an express workflow that runs for 30 seconds, the total cost = $0.057
If we repeat the calculation for an express workflow that runs for 1 second to process 1 object because you cannot utilize batching, the total cost = $13.42
What did you observe?
Express workflows are cheaper when the duration is lesser. They are also cost effective if there are more steps in the child workflow or your distributed map cannot make use of batching. Remember, Standard workflows are priced by state transitions meaning when number of steps and number of child workflow executions increase, cost increases.
You can look at from the below chart how express workflow duration affect the cost.
and
Navigate to
Each child workflow receives a batch of 100 objects. The concurrency or the parallelism of workflow is set as 1000.
CODE"ItemBatcher": {
"MaxItemsPerBatch": 100
},
"MaxConcurrency": 1000,
Navigate to
Save and Execute the workflow with default input
Explore the map run results. You can see 1000 child workflow executions.
All the child workflows are completed in little under 25 seconds
Repeat the exercise with different batch settings. What did you observe?
Yes, The total duration increases when you increase the batch size since a single Lambda is looping through an array passed to it, thus increasing the duration of Lambda execution.
Assume you need to process 50M objects. You have 2 steps inside your child workflow. Each child workflow processes 100 objects in a batch. The number of state transition per child workflow is 3. Total number of child workflows to process 50M objects is 500,000.
Total cost = (number of transitions per execution x number of executions) x $0.000025
Total number of child workflows to process 50M objects = 500,000
Total cost = (3 * 500000) x $0.000025 = $37.5
Here is a visual representation of the same information
Module 3 — Use Cases
Discovering and reporting vulnerabilities and security issues by scanning documents is a common process. If you are a security partner, when a new customer is onboarded, there can be hundreds of thousands of files to scan. Similarly, if a security procedure changes, previously scanned files may need to be rescanned. Scanning a large number of files is a both time consuming and expensive process. In this module, you will learn to scale a vulnerability scanning application to quickly and efficiently handle hundreds of thousands of files.
In the US healthcare system, claims are typically categorized as professional, institutional, or dental claims when they are submitted to health insurance payers. Health plans are responsible for validating these claims, responding to the provider, assessing the claims, making payments to the provider, and providing an explanation of benefits to the member. In this module, we focus on the validation phase of the claims process, which occurs after the claims data has already been converted to comply with the FHIR specification. During the validation phase, various business rules are applied to validate and enrich the claims. This represents the final step in the incoming flow of claims before they are transformed into custom data formats required by backend claims adjudication systems.
You will build a Step Functions workflow that processes healthcare claims data in a highly parallel fashion. The workflow uses the Distributed Map state that runs multiple child workflows, each processing a batch of the overall claims data. Each child workflow picks a set of individual claims files and processes them using table and then apply rules to determine validity of the claims. Upon processing the claims, the functions returns the output back to the workflow.
Learn how to use and configure a Distributed Map state for Healthcare claims data processing
Analyze the results of Distributed Map run
Challenge yourself to optimize the solution
— Object storage built to retrieve any amount of data from anywhere— Serverless compute service; Run code without thinking about servers or clusters
Data processing code in the following table (DMapHealthCareClaimTable).
DMapHealthCareRuleEngineLambdaFunction: This function reads data from the to simulate patient claims data in
Navigate to
- Select patterns tab and drag Process S3 objects onto the Workflow Studio canvas.
Select Start execution and use the default input payload.
The execution will take up to 5 minutes to complete successfully.
In the execution details page, select Distributed Map state in the Graph View, then select Details tab.
We can see that 21 child workflow executions completed successfully with 0 failures. Each child workflow processed 50 files.
We can view the duration of each child workflow execution. You can see overlapping timestamps for the start and end times, indicating that the data was processed in parallel.
If you select the execution name, you can use the Execution Input and output tab to view the input files for a child workflow execution and the execution output with details.
The Validate Claim function will apply the rules on the claims and store the claim status (Approved / Rejected) along with the rejected reason in the DynamoDB table DMapHealthCareClaimTable. You can use the gear icon on the right side of the screen to select which columns you want to view.
You have come to the end of the Healthcare Claim Processing Module. In this module, you created a Workflow with distributed map, learnt some important attributes of distributed map definition and run the workflow yourself.
Congratulations! You used Distributed Map state to quickly process a large dataset using parallel processing.
Extra Credits
Great! You have now executed and analyzed the results of the workflow. Well done!
But you cannot stop there! You will need to optimize for performance and cost!
Here is a list of things to try in order to understand the various handles you have at your disposal to optimize a workflow:
Increase the concurrency limit to 1000 and execute it again. Does it change the duration of the execution?
What happens if you decrease the Item Batching size to 25 and execute the workflow? What is the impact on duration as well as cost?
What combination of concurrency limit and batching size would be optimal?
What happens if you change the type of the workflow to ‘Express’ and execute it? What is the impact on cost? Would this workflow type work for any batching size of the provided data set?
Review what you learnt previously in this workshop on
You are developing a security vulnerability scanning application that alerts you of sensitive information in plain text files. The application you have built executes a workflow that scans a single file for exposed social security numbers (SSNs) and, if one is detected, sends a message to a queue for further downstream processing.
In this module, you will scale your security vulnerability scanning application using Step Functions Distributed Map to quickly address this backlog by processing multiple files concurrently.
— Visual workflows for distributed applications
— Fully managed message queuing for microservices, distributed systems, and serverless applications
Reviewing the Workflow
Open the
Select the Lambda state titled “Scan”.
Under API Parameters, click View function to review the Code.
Reference
.
Click the name of the bucket containing “vulnerabilitydatabucket”.
Copy both the full name of the bucket and the name of a file in that bucket.
Return to Workflow Studio.
Click Execute.
Enter the following input, replacing [bucket name] and [file name] with the names you copied:
{
"detail": {
"bucket": {
"name": "[bucket name]"
},
"object": {
"key": "[file name]"
}
}
}
Your successful graph will vary depending on the file name you copied. Only executions that process a file with an exposed SSN will send a message to the queue.
Scaling the Workflow with Distributed Map
In this section, you will add a Distributed Map state around the existing workflow to process multiple files concurrently.
Click Edit state machine.
Select the Code tab.
Overwrite the JSON definition with the following:
{
"StartAt": "Map",
"States": {
"Map": {
"Type": "Map",
"ItemProcessor": {
"ProcessorConfig": {
"Mode": "DISTRIBUTED",
"ExecutionType": "STANDARD"
},
"StartAt": "Scan",
"States": {
"Scan": {
"Type": "Task",
"Resource": "arn:aws:states:::lambda:invoke",
"OutputPath": "$.Payload",
"Parameters": {
"Payload.$": "$",
"FunctionName": ""
},
"Retry": [
{
"ErrorEquals": [
"Lambda.ServiceException",
"Lambda.AWSLambdaException",
"Lambda.SdkClientException",
"Lambda.TooManyRequestsException"
],
"IntervalSeconds": 2,
"MaxAttempts": 6,
"BackoffRate": 2
}
],
"Next": "Choice"
},
"Choice": {
"Type": "Choice",
"Choices": [
{
"Variable": "$.ssns",
"IsPresent": true,
"Next": "Queue"
}
],
"Default": "Pass"
},
"Queue": {
"Type": "Task",
"Resource": "arn:aws:states:::sqs:sendMessage",
"Parameters": {
"MessageBody.$": "$",
"QueueUrl": ""
},
"End": true
},
"Pass": {
"Type": "Pass",
"End": true
}
}
},
"End": true,
"Label": "Map",
"MaxConcurrency": 1000,
"ItemReader": {
"Resource": "arn:aws:states:::s3:listObjectsV2",
"Parameters": {
"Bucket.$": "$.bucket",
"Prefix.$": "$.prefix"
}
}
}
}
}
Select the Design tab.
- Click Start execution.
Again, your output will vary depending on the file that was executed on. Executions that process a clean file with no exposed SSN will return an empty object.
- Return to the parent execution view.
.
Select the queue containing “VulnerabilitiesQueue” in its name.
Click Send and receive messages.
Click Poll for messages.
Click a given message to view the Body.
The body of the message contains the location and serial number of the SSN(s).
- Click Done.
Notice the number of Messages available in the queue. You will now purge the queue in anticipation of the next full state machine execution.
- Click the queue name.
Return to the Step Functions execution page.
Click Edit state machine.
Select the Map state.
You can optimize the performance and cost of your workflow by selecting a batch size that balances the number of items against the items processing time. If you use batching, Step Functions adds the items to an Items array. It then passes the array as input to each child workflow execution.
Under Item batching, select Enable batching.
Set the Max MBs per batch to 50 KBs.
With batch input, you can pass a global JSON input to each child execution, merged with the inputs for items. In the last section, the bucket name was included in every single item, increasing the total input size. Instead of including the bucket name repeatedly with ItemSelector, you will include it once in the batch input.
Enter the following Batch input:
{
"bucket.$": "$.bucket"
}
Under Item source, expand Additional configuration.
Under Modify items with ItemSelector, overwrite the JSON with the following, removing details of the bucket:
{
"key.$": "$$.Map.Item.Value.Key"
}
Unselect Limit number of items to process all your files.
Express Workflows only run for up to five minutes and batch processing increases the number of files processed per child execution. If you configure a sufficiently large batch size, you may need to use a Standard Workflow.
Under Child execution type, select Standard.
Select the Code tab to review the JSON definition.
Your FunctionName and QueueUrl will be different.
CODE{
"StartAt": "Map",
"States": {
"Map": {
"Type": "Map",
"ItemProcessor": {
"ProcessorConfig": {
"Mode": "DISTRIBUTED",
"ExecutionType": "STANDARD"
},
"StartAt": "Scan",
"States": {
"Scan": {
"Type": "Task",
"Resource": "arn:aws:states:::lambda:invoke",
"OutputPath": "$.Payload",
"Parameters": {
"Payload.$": "$",
"FunctionName": "arn:aws:lambda:us-east-1:172187416625:function:vulnerability-scanning-mo-VulnerabilityScanningFun-G786d6ZNTaPI:$LATEST"
},
"Retry": [
{
"ErrorEquals": [
"Lambda.ServiceException",
"Lambda.AWSLambdaException",
"Lambda.SdkClientException",
"Lambda.TooManyRequestsException"
],
"IntervalSeconds": 2,
"MaxAttempts": 6,
"BackoffRate": 2
}
],
"Next": "Choice"
},
"Choice": {
"Type": "Choice",
"Choices": [
{
"Variable": "$.ssns",
"IsPresent": true,
"Next": "Queue"
}
],
"Default": "Pass"
},
"Queue": {
"Type": "Task",
"Resource": "arn:aws:states:::sqs:sendMessage",
"Parameters": {
"MessageBody.$": "$",
"QueueUrl": "https://sqs.us-east-1.amazonaws.com/172187416625/vulnerability-scanning-module-VulnerabilitiesQueue-UtFIvCLDH5Lh"
},
"End": true
},
"Pass": {
"Type": "Pass",
"End": true
}
}
},
"End": true,
"Label": "Map",
"MaxConcurrency": 1000,
"ItemReader": {
"Resource": "arn:aws:states:::s3:listObjectsV2",
"Parameters": {
"Bucket.$": "$.bucket",
"Prefix.$": "$.prefix"
},
"ReaderConfig": {}
},
"ItemSelector": {
"key.$": "$$.Map.Item.Value.Key"
},
"ItemBatcher": {
"MaxInputBytesPerBatch": 51200,
"BatchInput": {
"bucket.$": "$.bucket"
}
}
}
}
}
- If the changes to the JSON definition look accurate, click Save.
Overwrite the code with the following:
import json
import boto3
import re
def handler(event, context):
bucket = event["BatchInput"]["bucket"]
ssns = []
for item in event["Items"]:
CODEkey = item["key"]
obj = boto3.client('s3').get_object(
Bucket=bucket,
Key=key
)
body = obj['Body'].read().decode()
searches = re.findall("ssn=[^\s]+", body)
if searches:
ssns.extend([{"key": key, "serial": number[-4:]}
for ssn, number in (search.split("=") for search in searches)
])
if ssns:
return {"ssns": ssns}
else:
return {}
Click Deploy.
Executing the Batch Workflow
Return to Workflow Studio.
Click Execute.
Enter the following input, replacing [bucket name] with the bucket name you copied::
{
"bucket": "[bucket name]",
"prefix": ""
}
Under Events, click Map Run.
Wait a moment, if necessary, then click a given child execution.
After a successful invocation, expand the ID 6 dropdown to see the output of the Lambda function.
Return to the parent execution view.
.
Click the refresh icon, if necessary.
Learn how Step Functions Distributed Map can use a Step Functions Activity to distribute work to workers almost anywhere
Learn how Step Functions can manage workers through its built-in Amazon ECS integrations
Notice the number of Messages available in the queue. Since you removed the limitation on the number of items processed, it is much higher than before.
Summary
In this module, you added a Distributed Map state to your workflow to scale your security vulnerability scanning application across multiple files concurrently. By refactoring your application and enabling batch processing, you further optimized the performance and cost of your workflow.
A Monte Carlo simulation is a mathematical technique that allows us to predict different outcomes for various changes to a given system. In financial portfolio analysis the technique can be used to predict likely outcomes for aggregate portfolio across a range of potential conditions such as aggregate rate of return or default rate in various market conditions. The technique is also valuable in scenarios where your business case requires predicting the likely outcome of individual portfolio assets such detailed portfolio analysis or stress tests.
For this fictitious use case we will be working with a portfolio of personal and commercial loans owned by our company. Each loan is represented by a subset of data housed in individual S3 objects. Our company has tasked us with trying to predict which loans will default in the event of a Federal Reserve rate increase.
Loan defaults occur when the borrower fails to repay the loan. Predicting which loans in a portfolio would default in various scenarios helps companies understand their risk and plan for future events.
. Activities are an AWS Step Functions feature that enables you to have a task in your state machine where the work is performed by a worker that can be hosted on Amazon Elastic Compute Cloud (Amazon EC2), Amazon Elastic Container Service (Amazon ECS), mobile devices — basically anywhere. Think of activity as a Step Functions managed internal queue. You use an activity state to send data to the queue, one or more workers will consume the data from the queue. For this solution we utilize Amazon ECS on Amazon Fargate to run our Activity Workers (Worker).
to provides end to end orchestration for processing billions of records with your simulation or transformation logic using AWS Step Functions features. At the start of the workflow, Step Functions will scale the number of workers to a (configurable) predefined number. It then reads in the dataset and distributes metadata about the dataset in with an . Though the workers could potentially run almost anywhere so long as they had access to poll the Step Functions Activity and report SUCCESS/FAILURE back to Step Functions.
Updating the ECS Service — Part 1
The solution uses Amazon ECS to run the workers that handle the actual data processing. In this example we have created an ECS Service that will run a variable number of ECS Tasks, controlled by our Step Functions workflow. The workers run asynchronously from the Distributed Map, which uses an Activity to distribute the dataset. In this step you will configure that ECS Service to use a Task Definition that was predefined in CloudFormation. Let’s get started.
Important
A task definition is a blueprint for your application. It is a text file in JSON format that describes the parameters and one or more containers that form your application. You can learn more
Choose the ECS Cluster named “sfn-fargate-dataproc-xxxxxxxx” (note the similar Cluster named sfn-fargate-datagen-xxxxxxxx, please choose the one ending in dataproc)
Choose ECS Cluster
Click the Update Service button in the top right of the ECS Service page
Update Service
- Leave all other fields default. Scroll to the bottom of the page and click Update.
- In your AWS Console navigate to the AWS Step Functions Console by using the search bar in the upper left corner of your screen, typing “step functions” and clicking the Step Functions icon.
- Click the “Start execution” button to start the workflow.
- We can then monitor the process of the Step Function State Machine process from the execution status screen. Processing of the records in the simulated dataset takes just a few minutes.
That’s it! You have successfully updated your ECS Service to use a new Task Definition. Now lets run our Step Functions State Machine.
Executing the Workflow
Let’s go ahead and run the Step Function and then we will walk through each step as well as some optimizations for you to consider….
Console Navigation
Step Function Execution
The Processing DMap step is an AWS Step Functions Distributed Map step that reads the S3 Inventory manifest provided by the Parent Map and processes the referenced
to reduce costs and eliminate maintaining EC2 instances. Fargate provides us with AWS managed compute for scheduling our containers.
ECS Cluster / Capacity Provider
Choose the ECS Cluster named “sfn-fargate-dataproc-xxxxxxxx” (note the similar Cluster named sfn-fargate-datagen-xxxxxxxx, please choose the one ending in dataproc)
Choose ECS Cluster
Click the Update Service button in the top right of the ECS Service page
Update Service
Leave all other fields default. Scroll to the bottom of the page and click Update.
Now that you have updated the Service, lets run the State Machine again. If you need instructions please refer back to
Choose the State Machine named “sfn-fargate-dataproc-xxxxxxxx”
Step Function Selection
On each tab view the details of the execution and find the Duration field to see how long each execution required.
Find Duration
then empty both buckets containing “hellodmap” in the name.
Open the . Search for sfw-optimization. Empty both buckets.
Navigate to . Search for dmapworkshophealthcare. Empty the bucket.
Navigate to then select the stack with a name containing “vulnerability-scanning-module” (or with the name you entered earlier).
Click Delete.
Make sure the stack deletion completes.
Navigate to console.
Search for sfn-fargate. Delete the stack.
Search for sfn-datagen. Delete the stack.
Challenges Faced and Solutions
Challenge 1: Complex Workflow Management
Solution: Leveraged AWS Step Functions’ visual editor to design and troubleshoot each state transition, ensuring a seamless workflow.
Challenge 2: Detailed Monitoring Across Multiple Services
Solution: Integrated AWS CloudWatch and X-Ray to gain a comprehensive view of workflow execution, enabling more effective troubleshooting.
Challenge 3: Ensuring Data Security
Solution: Implemented strict IAM policies to ensure secure access control, preventing unauthorized access to sensitive data.
Conclusion
This project showcases how AWS Step Functions can effectively orchestrate complex data workflows across multiple services, enabling seamless scalability and enhanced reliability. By combining automated error handling, real-time monitoring, and optimized processing techniques, this architecture demonstrates a highly adaptable solution for data-driven organizations. The end result is a streamlined, resilient system capable of handling large datasets efficiently, supporting businesses in making data-informed decisions with minimal operational overhead.
Explore my ,
SOCIAL SHARE CARD GENERATOR