AWS Step Functions Batch Process

lambda-code.js

const _ = require('lodash');
const { mapSeries } = require('bluebird');

// express handler
async (req, res) => {
  const sfExecutionIdHeader = req.header('SF-Execution-Id');
  const executionId = sfExecutionIdHeader.substring(sfExecutionIdHeader.lastIndexOf(':') + 1);

  const ids = await selectIdsFromDb();
  const batches = _.chunk(ids, config.get('batch_size'));

  const batchS3Keys = await mapSeries(batches, async (batch, index) => {
    const key = `${executionId}/${index}/batch.json`;
      
    await s3Client.upload({ 
        Bucket: config.get('bucket'),
        Key: key,
        Body: JSON.stringify(batch)
    }).promise();

    return key;
  });

  res.status(200).send({
    batchS3Keys,
  });
}

template.yml

Definition:
  StartAt: PrepareBatches
  States:
    PrepareBatches:
      Type: Task
      Resource: arn:aws:states:::lambda:invoke
      Parameters:
        FunctionName: !Ref MyLambda
        Payload:
          # using nodejs express application inside the lambda
          httpMethod: POST
          path: /prepare-batches
          headers:
            SF-Execution-Id.$: $$.Execution.Id
      ResultSelector:
        body.$: States.StringToJson($.Payload.body)
        statusCode.$: $.Payload.statusCode
      Next: InvokeBatches

      InvokeBatches:
        Type: Map
        # previous lambda invocation writes all the S3 keys in this field
        ItemsPath: $.body.batchS3Keys
        MaxConcurrency: 1
        Iterator:
          StartAt: InvokeCaller
          States:
            InvokeCaller:
              Type: Task
              Resource: arn:aws:states:::states:startExecution.sync:2
              Parameters:
                StateMachineArn: !GetAtt CallerStepFunction.Arn
                Input:
                  batchS3Key.$: $
                  AWS_STEP_FUNCTIONS_STARTED_BY_EXECUTION_ID.$: $$.Execution.Id
              End: true
              Retry:
                - ErrorEquals: ["States.TaskFailed"]
                  IntervalSeconds: 10
                  MaxAttempts: 2
              Catch:
                - ErrorEquals: ["States.TaskFailed"]
                  Next: CallerFailed
            CallerFailed:
              Type: Succeed

        End: true