Single source pattern

In a single source pattern, Distributed Map reads data from an Amazon S3 object or JSON array which is given as an input to the map. It then fans out highly concurrent processing by distributing batches of the data to child workflow defined within the map. With each iteration, the child workflow runs as many parallel workflows as configured with MaxConcurrency attribute. The results of the processing are aggregated and stored in S3 bucket for further processing.

Many organizations use flat files to aggregate, store, and transfer data across disparate systems. For example, a third party partner sends pricing data across all the products periodically in .csv file format. You need to process the records in the file at scale so that the required changes are processed on time.

 "DistributedMap": {
      "Type": "Map",
      "ItemReader": {
        "ReaderConfig": {
          "InputType": "CSV",
          "CSVHeaderLocation": "FIRST_ROW"
        },
        "Resource": "arn:aws:states:::s3:getObject",
        "Parameters": {
          "Bucket.$": "$.bucket",
          "Key.$": "$.manifest"
        }
      } ...
  }

You create an AWS Lambda function that processes one or more of these records. You call the Lambda function inside the Distributed Map with a batch of records using ItemBatcher. You also control how many instances of Lambda functions are invoked in parallel using MaxConcurrency.

 "DistributedMap": {...
        "StartAt": "Lambda Invoke",
        "States": {
          "Lambda Invoke": {
            "Type": "Task",
            "Resource": "arn:aws:states:::lambda:invoke",
            "OutputPath": "$.Payload",
            "Parameters": {
              "Payload.$": "$",
              "FunctionName": "arn:aws:lambda:us-east-1:1234567890:function:priceupdate:12"
            },
            "Retry": [
              {
                "ErrorEquals": [
                  "Lambda.ServiceException",
                  "Lambda.AWSLambdaException",
                  "Lambda.SdkClientException",
                  "Lambda.TooManyRequestsException"
                ],
                "IntervalSeconds": 2,
                "MaxAttempts": 6,
                "BackoffRate": 2
              }
            ],
            "End": true
          }
        }
      },
      "MaxConcurrency": 1000,
      "ItemBatcher": {
        "MaxItemsPerBatch": 500
      },
      "Next": "Notifier"
    }

The file can have up to 100M records or 10GB in size. The file type supported are .csv and JSON.

Use cases

1. Payroll processing for millions of associates with records stored in .csv or JSON format.
2. Migrate data from one data store to another using a data export from S3.

To test out single source pattern in action, see the Distributed Map state gif generator.

Multi file pattern

In the multi file pattern, Distributed Map accepts S3 file prefix, collects metadata about files, and distributes the metadata to child workflow for further processing. With each iteration, the child workflow runs as many parallel workflows as configured with MaxConcurrency attribute. The results of the processing are aggregated and stored in an S3 bucket for further processing.

Consider an example where sensors in manufacturing units send operational metrics every 5 minutes. You have as many files as the number of machines in the 5 minute period. At every 5 minute, you run am AWS Step Functions workflow that process all the files, identify anomalies, and take immediate action.

    "DistributedMap": {
      "Type": "Map",
      "ItemReader": {
        "Resource": "arn:aws:states:::s3:listObjectsV2",
        "Parameters": {
          "Bucket": "$.sensor_data_bucket"
          "Prefix.$": "$.fivemin_data_prefix"
        }
      }...

The resource provided in this pattern uses the listObjectsV2 API whereas the single file pattern uses getObject. You write a Lambda function to run the calculation to detect anomalies.

        "States": {
          "Lambda Invoke": {
            "Type": "Task",
            "Resource": "arn:aws:states:::lambda:invoke",
            "OutputPath": "$.Payload",
            "Parameters": {
              "Payload.$": "$",
              "FunctionName": "arn:aws:lambda:us-east-1:1234567890:function:process-sensordata--C1x7u12ryFZ2"
            }....
        }
 

The Lambda function in the child workflow receives one or more file metadata read from the configured bucket/prefix location.

{
  "Items": [
    {
      "Etag": "\"6653728b15bff8bd5d33ebd492500273\"",
      "Key": "98430099999_sensor1.xml",
      "LastModified": 1675979708,
      "Size": 8821,
      "StorageClass": "STANDARD"
    },
    {
      "Etag": "\"ee80f36288e29747af43d081dc9d05a0\"",
      "Key": "98431099999_sensor2.xml",
      "LastModified": 1675979708,
      "Size": 9391,
      "StorageClass": "STANDARD"
    }
    
 }

Use cases

• Scan thousands of files for security vulnerabilities .
• Run an ML model inferencing to predict financial risk scores for your customers with input data stored in individual files in S3.
• Backfill data transformation for past dates when you onboard a new customer.

To test out the multi file pattern, see the Large-Scale Parallelization with Distributed Map workshop. This is a step-by-step example of how to use a Step Functions workflow in Distributed mode to process a list of S3 objects in an S3 bucket.

Nested single source pattern

In a nested single source pattern, Distributed Map is inside another Distributed Map. The outer map reads metadata of multiple files from an S3 bucket, and sends batches of metadata to the inner map. Inner map reads the individual file records and runs child workflows to process them. The data can be aggregated at each Distributed Map for further processing.

Imagine you are getting end of day sales records from point of sales systems across your restaurants. Each individual files has 10s of 1000s of sales data for a single restaurant. You need to clean, verify, process payments, and generate a report.

    "OuterMap to iterate sales across restaurants": {
      "Type": "Map",
      "ItemProcessor": {
        "ProcessorConfig": {
          "Mode": "DISTRIBUTED",
          "ExecutionType": "STANDARD"
        },
        "StartAt": "InnerMap to iterate sales records",
        "States": {
          "InnerMap to iterate sales records": {
            "Type": "Map",
            "ItemProcessor": {
              "ProcessorConfig": {
                "Mode": "DISTRIBUTED",
                "ExecutionType": "EXPRESS"
              },...

The outer map iterates through each S3 objects located at the prefix using S3 listObject, sends the file metadata to the child workflow which is a Distributed Map itself.

      "ItemReader": {
        "Resource": "arn:aws:states:::s3:listObjectsV2",
        "Parameters": {
          "Bucket": "endofday_sales",
          "Prefix.$": "$.day"
        }

The inner map reads a single S3 object from the outer map, iterates on the content which is sales records, and sends it to the Lambda function in the child workflow.

            "ItemReader": {
              "ReaderConfig": {
                "InputType": "CSV",
                "CSVHeaderLocation": "FIRST_ROW"
              },
              "Resource": "arn:aws:states:::s3:getObject",
              "Parameters": {
                "Bucket": "endofday_sales",
                "Key.$": "$.Key"
              }
            }

Use cases

• End of day financial transactions from multiple locations that need to be processed, verified for fraud, reviewed manually by agent if needed.
• Process thousands of invoices each consisting of 100s of sales records.