Problem Let's consider such a case — you are developing functionality that processes thousands of records in any data storage and your weapon of choice is the AWS Step Function service. One of the most powerful tools you can use there is Map task type since there you can maximize your function load and process chunks of data concurrently with param. maxConcurrency Let’s assume the logic of the steps: CaptureInput — will accept some params from the user/service fetch data from storage and pass it to be processed ProcessData — consume data process it and update/delete whatever logic requires We also should keep in mind the main limitation of Inline Map for the step function Max payload size — 256 KB (whole payload, not just one array item) Max execution history — 25,000 events Accepts only a JSON array as input. Code An obvious part of the work that the Lamda should do is to fetch and pass some data into the Map state in the form JSON Array: CaptureInput from-to // capture-input/handler.js
exports.handler = async ({number = 10, chunk = 1}) => {

  return paginate(number, chunk)
}

function paginate(number, chunk){
  const result = [];
  const numPages = Math.ceil(number / chunk);
  for (let i = 0; i < numPages; i++) {
    result.push({
      from: i * chunk,
      to: Math.min((i + 1) * chunk, number),
    });
  }
  //  [{from: 1, to: 2}, {from: 2, to: 3}.....]
  return result;
}j For simplicity, imagine we are not passing raw ids but only a form — to cursor for a chunk of data. Lambda receives the following params: number — the number of processed data items chunk — count of the size that each map would process The Lambda would log it and somehow process the data: ProcessData // process-output/handler.js
exports.handler = (event) => {
  console.log(JSON.stringify(event))
  // ...do some processing
} For deploying the step function the plugin is used, the YAML file would look like this: serverless-step-function # serverless.yml
service: limit-payload-step-function

provider:
  name: aws
  runtime: nodejs18.x
  stage: dev
  region: us-east-1
  accountId: 670913836060
  architecture: arm64

package:
  individually: true

plugins:
  - serverless-step-functions

functions:
  captureInput:
    handler: capture-input/handler.handler
    name: ${self:provider.stage}-${self:service}-captureInput
    timeout: 900
    memorySize: 1024
    tags:
      name: captureInput

  processOutput:
    handler: process-output/handler.handler
    name: ${self:provider.stage}-${self:service}-processOutput
    timeout: 900
    memorySize: 128
    tags:
      name: processOutput

stepFunctions:
  stateMachines:
    ProcessDataStateMachine:
      name: process-data-state-machine
      definition:
        StartAt: CaptureInput
        States:
          CaptureInput:
            Type: Task
            Resource: arn:aws:lambda:${self:provider.region}:${self:provider.accountId}:function:${self:provider.stage}-${self:service}-captureInput
            Next: MapData
          MapData:
            MaxConcurrency: 5
            Type: Map
            InputPath: $
            Iterator:
              StartAt: ProcessData
              States:
                ProcessData:
                  Type: Task
                  Resource: arn:aws:lambda:${self:provider.region}:${self:provider.accountId}:function:${self:provider.stage}-${self:service}-processOutput
                  End: true
            End: true Then run to deploy: serverless deploy --aws-profile=personal Great! We have our function up and running and everything seems fine: Until we try some bigger input to be passed to Map: Let’s try running it with: {
  "number": 10000,
  "chunk": 5
} And now we are in trouble: * exceeding the maximum number of bytes service limit**.* The state/task returned a result with a size The output of is bigger than the step function can process (256 KB). Even if we try somehow to optimize params passing we still can hit 25k execution events when we would need to add other steps into the MapData task. And that is the time for to rescue us! CaptureInput Distributed Map Inline VS Distributed There is a neat illustration from AWS that outlines the work of the we used before: Inline Map Runs a set of steps for each element in a JSON array passed from the previous state. Step Functions adds the execution history of each Map iteration to the state machine’s execution history. Map iterations run in parallel, with limited concurrency after 40 iterations. As we can see, we are pretty much limited using the default Map, and that's why the AWS team created a that would overcome the limits of the previous one. New Distributed Mode Runs a set of steps for each item in a JSON array, S3 object list, or CSV file. Map iterations are fanned out as separate child executions with their own execution history, enabling very high concurrency. Iterates over and processes millions of items by running thousands of parallel child executions. Can cascade multiple Distributed Map states and batch items to create map-reduce type jobs. A Distributed Map state launches a Map Run resource, which you can access through the console, APIs, and SDK. As you can see, basically every one of these items is going to make work with larger datasets easier and faster. By checking out Step Function Studio, you can see that there are 3 main types of Distributed Map: In this case, I guess the most suitable is the second option: Let Map iterate over JSON that is stored in S3. This is an example of how the step function can be realized using this approach: Lambda receives params and generates from — to pais CaptureInput Saves JSON with the pairs into Bucket Pass the Key (path) of JSON and Bucket name to Distributed Map will automatically prefetch JSON and loop over each item in it, the item in the array is going to be passed directly into the Map body ( ) Map CaptureInput receive the item of JSON array and process it like input for further data manipulations. CaptureInput Coding the solution Creating bucket As was shown in the diagram above, we require AWS S3 Bucket to store items for Map. You can do it manually via the console or by adding section in file: Resource serverless.yml resources:
  Resources:
    PayloadBucket:
      Type: AWS::S3::Bucket
      DeletionPolicy: Retain
      Properties:
        BucketName: step-function-payloads Adding S3 permission Then since function required S3 permission, I'm using plugin to do that, but you can also do it manually. CaptureInput serverless-iam-role-per-function yarn add serverless-iam-roles-per-function Add it to the plugins section: provider:
  # .....
  environment:
    BUCKET_NAME: step-function-payloads
  iam:
    role:
      statements:
        - Effect: 'Allow'
          Action:
            - 's3:*'
          Resource:
            - 'arn:aws:s3:::step-function-payloads/*'
            - 'arn:aws:s3:::step-function-payloads'
# .....
plugins:
  - serverless-step-functions
  - serverless-iam-roles-per-function
# .....
custom:
  serverless-iam-roles-per-function:
    defaultInherit: true This config will attach role with access to to each function as a default. Also, it will help create granular permissions in further development. iam step-function-payload Saving Iterable JSON Now we need to write our output JSON into S3 so that later Map state will iterate over it. It will be done in lambda: CaptureInput // capture-input/handler.js
const AWS = require('aws-sdk')
const s3 = new AWS.S3();
const payloadFolder = 'temp-sf-payload'
const Bucket = process.env.BUCKET_NAME;

exports.handler = async ({number = 10, chunk = 1}) => {

  const data = paginate(number, chunk)

  const path = `${payloadFolder}/${Date.now()}.json`

  const res = await s3.putObject({
    Bucket,
    Key: path,
    Body: JSON.stringify(data)
  }).promise()

  console.log('Saved output: ', {Bucket, path})

  return {bucket: Bucket, path}
}

function paginate(number, chunk){
  const result = [];
  const numPages = Math.ceil(number / chunk);
  for (let i = 0; i < numPages; i++) {
    result.push({
      from: i * chunk,
      to: Math.min((i + 1) * chunk, number),
    });
  }

  return result;
} We just save the payload into an s3 file and pass and to the Map State. Now we need to consume it! from-to[] bucket path Distributed Map Definition There are not many changes in a Step function definition: # ...
MapData:
  Type: Map
  MaxConcurrency: 5
  ItemReader:
    Resource: arn:aws:states:::s3:getObject
    ReaderConfig:
      InputType: JSON
    Parameters:
      # Assign output of CaptureInput lambda for Map to iterate over
      Bucket.$: $.bucket
      Key.$: $.path
  ItemProcessor:
    ProcessorConfig:
      Mode: DISTRIBUTED
      ExecutionType: STANDARD
    StartAt: ProcessData
    States:
      ProcessData:
        Type: Task
        Resource: arn:aws:lambda:${self:provider.region}:${self:provider.accountId}:function:${self:provider.stage}-${self:service}-processOutput
        End: true
  End: true There are two new key components here: used to specify the source of iteration and its type - either JSON CSV or S3 Folder ItemReader — the same as in inline mode but with specifying nested workflow type ItemProcessor Iterator I’ve added all the references regarding the new Distributed mode including the blueprints repository in the last section of the article. Step function Role The last step is to add a custom role to the step function. New Map mode requires Map state to have getObject permissions to read from bucket we specified. resources:
  Resources:
    ProcessDataStateMachineStepFunctionRole:
      Type: "AWS::IAM::Role"
      Properties:
        RoleName: ${self:provider.stage}-ProcessDataStateMachine-StepFunctionRole
        AssumeRolePolicyDocument:
          Statement:
            - Effect: Allow
              Principal:
                Service:
                  - states.amazonaws.com
              Action:
                - sts:AssumeRole
        Policies:
          # Allow invoking lambdas as a tasks in step function
          - PolicyName: ${self:provider.stage}-${self:service}-lambda-invoke
            PolicyDocument:
              Version: '2012-10-17'
              Statement:
                - Effect: 'Allow'
                  Action:
                    - 'lambda:*'
                  Resource:
                    - 'arn:aws:lambda:*:*:function:${self:provider.stage}-*'
          # Allow starting execution
          - PolicyName: ${self:provider.stage}-${self:service}-start-execution
            PolicyDocument:
              Version: '2012-10-17'
              Statement:
                - Effect: 'Allow'
                  Action:
                    - 'states:StartExecution'
                  Resource:
                    - arn:aws:states:${self:provider.region}:${aws:accountId}:stateMachine:${self:stepFunctions.stateMachines.ProcessDataStateMachine.name}
          # Allow logging
          - PolicyName: ${self:provider.stage}-${self:service}-default-logging
            PolicyDocument:
              Version: '2012-10-17'
              Statement:
                - Effect: 'Allow'
                  Action:
                    - 'logs:CreateLogGroup'
                    - 'logs:CreateLogStream'
                    - 'logs:PutLogEvents'
                  Resource:
                    - 'arn:aws:logs:*:*:log-group:/aws/lambda/*'

                - Effect: 'Allow'
                  Action:
                    - 'xray:PutTraceSegments'
                    - 'xray:PutTelemetryRecords'
                  Resource:
                    - '*'
          # Full access to S3 bucket
          - PolicyName: ${self:provider.stage}-${self:service}-read-s3-bucket
            PolicyDocument:
              Version: '2012-10-17'
              Statement:
                - Effect: 'Allow'
                  Action:
                    - 's3:*'
                  Resource:
                    - arn:aws:s3:::step-function-payloads/* Demo Started function with these params: {
  "number": 100000,
  "chunk": 5
} Map successfully started! Now we can monitor the progress in a separate menu: This Map is processing 20k steps in a snap of a finger, and it’s not the last feature of Distributed mode. We also can increase and use parameters to make it even faster! MaxConcurrency ItemBatcher Summary Now you know how to overcome the limitations of the Inline Map of your step function workflow with Distributed Map task state. Check out the blueprint in the section below! Thanks for reading! Resources Repository: Step Function Distributed Map Example DOC: Map state processing modes DOC: Using Map state in Distributed mode DOC: ItemBatcher Originally published here

Too Long; Didn't Read

Developers: Build Less. Deliver More. [Learn how]	

Achieving Next-level Performance With Distributed Map!

Too Long; Didn't Read

Dmytro Harazdovskiy

About Author

TOPICS

THIS ARTICLE WAS FEATURED IN...

Achieving Next-level Performance With Distributed Map!

Too Long; Didn't Read

Dmytro Harazdovskiy

About Author

TOPICS

THIS ARTICLE WAS FEATURED IN...

RELATED STORIES