Using Map state in Distributed mode

Use the ItemReader field to specify the location of the dataset from which the Map state reads its input data. The following example shows how to use a CSV file as a dataset by specifying the Amazon S3 bucket name and object key in which it's stored.

{
  "Map": {
    "Type": "Map",
    ...
    "ItemReader": {
      "ReaderConfig": {
        "InputType": "CSV",
        "CSVHeaderLocation": "FIRST_ROW"
      },
      "Resource": "arn:aws-cn:states:::s3:getObject",
      "Parameters": {
        "Bucket": "Database2022",
        "Key": "csv-dataset/ratings.csv"
      }
    },
    ...
    "End": true
  }
}

Use the MaxItems sub-field of the ItemReader field to limit the number of items the Map state can read from a dataset. For example, if you specify the MaxItems sub-field value as 90 for a CSV dataset, the Map state only reads the first 90 rows of your CSV file, starting after the header row.

{
  "Map": {
    "Type": "Map",
    ...
    "ItemReader": {
      "ReaderConfig": {
        "MaxItems": 90,
        "InputType": "CSV",
        "CSVHeaderLocation": "FIRST_ROW"
      },
      ...
    },
    ...
    "End": true
  }
}

If your dataset is a JSON input passed from a previous step in the workflow and it contains an array, use the ItemsPath field to select the node that contains the array.

For example, given the following JSON input:

{
  "facts": [
    {
      "verdict": "true",
      "statement_date": "6/11/2008",
      "statement_source": "speech"
    },
    {
      "verdict": "false",
      "statement_date": "6/7/2022",
      "statement_source": "television"
    },
    {
      "verdict": "mostly-true",
      "statement_date": "5/18/2016",
      "statement_source": "news"
    }
  ]
}

Use the ItemsPath field as shown in the following example to select the JSON node that contains the array:

"Map": {
  "Type": "Map",
  "ItemProcessor": {
    "ProcessorConfig": {
      "Mode": "DISTRIBUTED",
      ...
    }
  },
  ...
  "ItemsPath": "$.facts",
  ...
  "End": true
}

Use the ItemSelector field to override the individual dataset items' values before they’re passed to the Map state iterations. To override the values, you specify a JSON input that contains a collection of key-value pairs. These pairs can either be static values that you provide in your state machine definition, values selected from the state input using a path, or values accessed from the context object.

For example, the following custom JSON input replaces the original input before it's passed on to the Map state iterations. Each child workflow execution receives the following custom input containing a static value and two context object data items.

{
  "Map": {
    "Type": "Map",
    ...
    "ItemSelector": {
      "foo": "bear",
      "index.$": "$$.Map.Item.Index",
      "value.$": "$$.Map.Item.Value"
    },
    ...
    "End": true
  }
}

By default, the Map state passes each item in the dataset as input to individual child workflow executions. Use the ItemBatcher field to process a group of items in each child workflow execution. In this field, you can specify the maximum number of items to batch, the maximum batch size in bytes, or both these values. The interpreter adds the specified number of items to an Items array. It then passes the array as input to each child workflow execution. For example, if you specify the maximum items to batch as 10, the interpreter adds the specified number of items to an Items array in the input to each child workflow execution.

The following example shows how to batch 10 dataset items by specifying the MaxItemsPerBatch field:

{
  "Map": {
    "Type": "Map",
    "ItemBatcher": {
      "MaxItemsPerBatch": 10
    },
    ...
    "End": true
  }
}

The following example shows a batch of items inside the Items array that the Map state passes as input to a child workflow execution:

{
  "Items": [
    {
      "rating": "3.0",
      "movieId": "1244",
      "userId": "2",
      "timestamp": "1192913551"
    },
    {
      "rating": "4.5",
      "movieId": "1296",
      "userId": "2",
      "timestamp": "1192913608"
    },
    ...
  ]
}

Use the ResultWriter field to export the results of all child workflow executions to an Amazon S3 bucket. Exporting the results to an Amazon S3 bucket is helpful if your output payload size exceeds 256 KB. To export the results of all child workflow executions, specify the Amazon S3 bucket name and object prefix where you want to store the results.

{
  "Map": {
    "Type": "Map",
    ...
    "ResultWriter": {
      "Resource": "arn:aws-cn:states:::s3:putObject",
      "Parameters": {
        "Bucket": "processedOutput",
        "Prefix": "test-run"
      }
    },
    ...
    "End": true
  }
}