Post Syndicated from Biswanath Mukherjee original https://aws.amazon.com/blogs/compute/optimizing-nested-json-array-processing-using-aws-step-functions-distributed-map/

When you’re working with large datasets, you’ve likely encountered the challenge of processing complex JSON structures in your automated workflows. You need to preprocess arrays within nested JSON objects before you can run parallel processing on them. Extracting data used to require custom code and extra processing steps, delaying you from building your core application logic.

With AWS Step Functions Distributed Map, you can process large datasets with concurrent iterations of workflow steps across data entries. Using the enhanced ItemsPointer feature of Distributed Maps, you can extract array data directly from JSON objects stored in Amazon S3. Alternatively, for JSON object as state input, you can use Items (JSONata) or ItemsPath (JSONPath). With this enhancement you can point directly to arrays nested within JSON structures, eliminating the need for custom preprocessing of your data. With ItemsPointer, Items, and ItemsPath you can select the nested array data and simplify your workflows.

In this post, we explore how to optimize processing array data embedded within complex JSON structures using AWS Step Functions Distributed Map. You’ll learn how to use ItemsPointer to reduce the complexity of your state machine definitions, create more flexible workflow designs, and streamline your data processing pipelines—all without writing additional transformation code or AWS Lambda functions.

Use case: e-commerce product data enrichment

In this e-commerce use case example, you’ll build a sample application that demonstrates processing of product inventory data for an e-commerce application using AWS Step Functions Distributed Map. The application receives a JSON file from an upstream application containing an array of product information. The Step Functions workflow reads the JSON file containing product data from an S3 bucket and iterates over the array to enrich each product data in the array.

The following diagram presents the AWS Step Functions state machine.

JSON array processing workflow

The JSON array is processed using the following workflow:

1. The state machine reads the product-updates.json file from an input S3 bucket. The file contains a JSON array of products.
2. The Distributed Map state in the state machine, selects the JSON array node using ItemsPointer and iterates over the JSON array.
3. For each of the items within the array, the state machine invokes a Lambda function for data enrichment. The Lambda function adds product stock and price information to the product data.
4. The state machine saves the updated product data in an Amazon DynamoDB table.
5. Finally, the state machine uploads the execution metadata into an output S3 bucket. See limits related to state machine executions and task executions.

MaxConcurrency can be configured to specify the number of child workflow executions in a Distributed Map that can run in parallel. If not specified, then Step Functions doesn’t limit concurrency and runs 10,000 parallel child workflow executions.

You can read a JSON file from a S3 bucket using ItemReader and its sub-fields. If the JSON file, from the S3 bucket, contains a nested object structure, you can select the specific node with your data set with an ItemsPointer. For example, the following input JSON file:

{
  "version": "2024.1",
  "timestamp": "2025-09-26T10:49:36.646197",
  "productUpdates": {
    "items": [
      {
        "productId": "PROD-001",
        "name": "Wireless Headphones",
        "price": 79.99,
        "stock": 150,
        "category": "Electronics"
      },
      {
        "productId": "PROD-002",
        "name": "Smart Watch",
        "category": "Electronics"
      },
      …
    ]
  }
}

The following JSONata-based workflow configuration extracts a nested list of products from productUpdates/items:

"ItemReader": {
   "Resource": "arn:aws:states:::s3:getObject",
   "ReaderConfig": {
      "InputType": "JSON",
      "ItemsPointer": "/productUpdates/items"
   },
   "Arguments": {
      "Bucket": "amzn-s3-demo-bucket",
      "Key": "updates/product-updates.json"
   }
}

For JSONPath-based workflow note that Arguments is replaced with Parameters:

"ItemReader": {
   "Resource": "arn:aws:states:::s3:getObject",
   "ReaderConfig": {
      "InputType": "JSON",
      "ItemsPointer": "/productUpdates/items"
   },
   "Arguments": {
      "Bucket": "amzn-s3-demo-bucket",
      "Key": "updates/product-updates.json"
   }
}

The ItemReader field is not needed when your dataset is JSON data from a previous step. ItemsPointer is only applicable when the input JSON objects read from an S3 bucket. If you are using JSON as state input to a Distributed Map, then you can use the ItemsPath (for JSONPath) or Items (for JSONata) field to specify a location in the input that points to JSON array or object used for iterations.

Prerequisite

To use Step Functions Distributed Map, verify you have:

• Access to an AWS account through the AWS Management Console and the AWS Command Line Interface (AWS CLI). The AWS Identity and Access Management (IAM) user that you use must have permissions to make the necessary AWS service calls and manage AWS resources mentioned in this post. While providing permissions to the IAM user, follow the principle of least-privilege.
• AWS CLI installed and configured. If you are using long-term credentials like access keys, follow manage access keys for IAM users and secure access keys for best practices.
• Git Installed
• AWS Serverless Application Model (AWS SAM) installed
• Python 3.13+ installed

Set up and run the workflow

Run the following steps to deploy the Step Functions state machine.

1. Clone the GitHub repository in a new folder and navigate to the project folder.

   git clone https://github.com/aws-samples/sample-stepfunctions-json-array-processor.git
   cd sample-stepfunctions-json-array-processor

2. Run the following commands to deploy the application.

   sam deploy --guided

   Enter the following details:

   • Stack name: Stack name for CloudFormation (for example, stepfunctions-json-array-processor)
   • AWS Region: A supported AWS Region (for example, us-east-1)
   • Accept all other default values.

   The outputs from the sam deploy will be used in the subsequent steps.

3. Run the following command to generate product-updates.json file containing a nested JSON array of sample products and upload the product-updates.json file to the input S3 bucket. Replace InputBucketName with the value from sam deploy output.

   python3 scripts/generate_sample_data.py <InputBucketName>

4. Run the following command to start execution of the Step Functions workflow. Replace the StateMachineArn with the value from sam deploy output.

   aws stepfunctions start-execution \
     --state-machine-arn <StateMachineArn> \
     --input '{}'

   The state machine reads the input product-updates.json file and invokes a Lambda function to update the database for every product in the array after adding price and stock information. The execution metadata is also uploaded into the results bucket.

Monitor and verify results

Run the following steps to monitor and verify the test results.

1. Run the following command to get the details of the execution. Replace executionArn with your state machine ARN.

   aws stepfunctions describe-execution --execution-arn <executionArn>

   Wait until the status shows SUCCEEDED.

2. Run the following commands to validate the processed output from ProductCatalogTableName DynamoDB table. Replace the value ProductCatalogTableName with the value from sam deploy output.

   aws dynamodb scan --table-name <ProductCatalogTableName>

3. Check that the DynamoDB table contains the enriched product data including price and stock attributes. Example output:

   {
       "Items": [
           {
               "ProductId": {
                   "S": "PROD-005"
               },
               "lastUpdated": {
                   "S": "2025-10-07T20:33:34.507Z"
               },
               "stock": {
                   "N": "129"
               },
               "price": {
                   "N": "139.25"
               }
           },
           {
               "ProductId": {
                   "S": "PROD-003"
               },
               "lastUpdated": {
                   "S": "2025-10-07T20:33:34.576Z"
               },
               "stock": {
                   "N": "471"
               },
               "price": {
                   "N": "40.92"
               }
           },
   	      …
       ],
       "Count": 5,
       "ScannedCount": 5,
       "ConsumedCapacity": null
   }

Clean up

To avoid costs, remove all resources you’ve created while following along with this post.

Run the following command after replacing the <placeholder> variable to delete the resources you deployed for this post’s solution:

aws s3 rm s3://<InputBucketName> --recursive
aws s3 rm s3://<ResultBucketName> --recursive
sam delete

Conclusion

In this post, you learned how to use Step Functions Distributed Map for extracting array data natively from JSON objects stored in a S3 bucket. By removing custom data extraction code, you can simplify the processing of your large-scale parallel workloads. With ItemsPointer you can extract array data within JSON files stored in a S3 bucket , and with Items(JSONata) or ItemsPath (JSONPath), you can extract arrays from complex JSON state input, adding flexibility to your workflow designs.

New input sources for Distributed Map are available in all commercial AWS Regions where AWS Step Functions is available. For a complete list of AWS Regions where Step Functions is available, see the AWS Region Table. To get started, you can use the Distributed Map mode today in the AWS Step Functions console. To learn more, visit the Step Functions developer guide.

For more serverless learning resources, visit Serverless Land.

Post Syndicated from Biswanath Mukherjee original https://aws.amazon.com/blogs/compute/processing-amazon-s3-objects-at-scale-with-aws-step-functions-distributed-map-s3-prefix/

If you’re building large scale enterprise applications, you’ve likely faced the complexities of processing large volumes of data files. Whether you’re analyzing your application logs, processing customer data files, or transforming machine learning datasets, you know the complexity involved in orchestrating workflows. You’ve probably written nested workflows and additional custom code to process objects from Amazon Simple Storage Service (Amazon S3) buckets.

With AWS Step Functions Distributed Map, you can process large scale datasets by running concurrent iterations of workflow steps across data entries in parallel, achieving massive scale with simplified management.

With the new prefix-based iteration feature and LOAD_AND_FLATTEN transformation parameter for Distributed Map, your workflows can now iterate over S3 objects under a specified prefix using S3ListObjectsV2 to process their contents in a single Map state, avoiding nested workflows and reducing operational complexity.

In this post, you’ll learn how to process Amazon S3 objects at scale with the new AWS Step Functions Distributed Map S3 prefix and transformation capabilities.

Use case: Application log processing and summarization

You’ll build a sample Step Functions state machine that demonstrates processing of all the log files from the given S3 prefix using a Distributed Map. You’ll analyze all the log files to build a summary INFO, WARNING and ERROR messages in the log file on hourly basis. The following diagram presents the AWS Step Functions state machine:

Log files processing workflow

1. The state machine iterates over all the log files from the specified S3 prefix using S3 ListObjectsV2 and process them using AWS Step Functions Distributed Map.
2. For each log file entry, the state machine puts hourly ErrorCount metric into Amazon CloudWatch.
3. The state machine then stores hourly metrics count in a Amazon DynamoDB table.
4. The state machine then invokes an AWS Lambda function to perform metrics aggregation.

The following is an example of the parameters in an ItemReader configured to iterate over the content of S3 objects using S3 ListObjectsV2.

{
  "QueryLanguage": "JSONata",
  "States": {
    ...
    "Map": {
        ...
        "ItemReader": {
            "Resource": "arn:aws:states:::s3:listObjectsV2",
            "ReaderConfig": {
                // InputType is required if Transformation is LOAD_AND_FLATTEN. Use one of the given values
                "InputType": "CSV | JSON | JSONL | PARQUET",
                // Transformation is OPTIONAL and defaults to NONE if not present
                "Transformation": "NONE | LOAD_AND_FLATTEN" 
            },
            "Arguments": {
                "Bucket": "amzn-s3-demo-bucket1",
                "Prefix": "{% $states.input.PrefixKey %}"
            }
        },
        ...
    }
}

With the LOAD_AND_FLATTEN option, your state machine will do the following:

1. Read the actual content of each object listed by the Amazon S3 ListObjectsV2 call.
2. Parse the content based on InputType (CSV, JSON, JSONL, Parquet).
3. Create items from the file contents (rows/records) rather than metadata.

We recommend including a trailing slash on your prefix. For example, if you select data with a prefix of folder1, your state machine will process both folder1/myData.csv and folder10/myData.csv. Using folder1/ will strictly process only one folder. All of the objects listed by prefix need to be in the same data format. For example, if you are selecting InputType as JSONL, your S3 prefix should contain only JSONL files and not a mix of other types.

The context object is an internal JSON structure that is available during an execution. The context object contains information about your state machine and execution. Your workflows can reference the context object in a JSONata expression with $states.context.

Within a Map state, the context object includes the following data:

"Map": {
   "Item": {
      "Index" : Number,
      "Key"   : "String", // Only valid for JSON objects
      "Value" : "String",
      "Source": "String"
   }
}

For each Map iteration, the Index contains the index number for the array item that is being currently processed.

A Key is only available when iterating over JSON objects. Value contains the array item being processed. For example, for the following input JSON object, Names will be assigned to Key and {"Bob", "Cat"} will be assigned to Value.

{
	"Names": {"Bob", "Cat"}
} 

Source contains one of the following:

• For state input: STATE_DATA
• For Amazon S3 LIST_OBJECTS_V2 with Transformation=NONE, the value will show the S3 URI for the bucket. For example: S3://amzn-s3-demo-bucket1
• For all the other input types, the value will be the Amazon S3 URI. For example: S3://amzn-s3-demo-bucket1/object-key

Using LOAD_AND_FLATTEN and the Source field, you can connect child executions to their sources.

Prerequisites

• Access to an AWS account through the AWS Management Console and the AWS Command Line Interface (AWS CLI). The AWS Identity and Access Management (IAM) user that you use must have permissions to make the necessary AWS service calls and manage AWS resources mentioned in this post. While providing permissions to the IAM user, follow the principle of least-privilege.
• AWS CLI installed and configured. If you are using long-term credentials like access keys, follow manage access keys for IAM users and secure access keys for best practices.
• Git Installed.
• AWS Serverless Application Model (AWS SAM) installed.
• Python 3.13 or later installed.

Set up and run the workflow

Run the following steps to deploy and test the Step Functions state machine.

1. Clone the GitHub repository in a new folder and navigate to the project folder.

   git clone https://github.com/aws-samples/sample-stepfunctions-s3-prefix-processor.git
   cd sample-stepfunctions-s3-prefix-processor

2. Run the following commands to deploy the application.

   sam deploy --guided

3. Enter the following details:
   • Stack name: Stack name for CloudFormation (for example, stepfunctions-s3-prefix-processor)
   • AWS Region: A supported AWS Region (for example, us-east-1)
   • Accept all other default values.

   The outputs from the AWS SAM deploy will be used in the subsequent steps.

4. Run the following command to generate sample log files.

   python3 scripts/generate_logs.py

5. Run the following to upload the log files to the S3 bucket with the /logs/daily prefix. Replace amzn-s3-demo-bucket1 with the value from the sam deploy output.

   aws s3 sync logs/ s3://amzn-s3-demo-bucket1/logs/ --exclude '*' --include '*.log'

6. Run the following command to execute the Step Functions workflow. Replace the StateMachineArn with the value from the sam deploy output.

   aws stepfunctions start-execution \
     --state-machine-arn <StateMachineArn> \
     --input '{}'

   The Step Function state machine iterates over all the log files with the S3 prefix /logs/daily and processes them in parallel. The workflow updates the metrics in CloudWatch, stores hourly metrics count in DynamoDB, then invokes an AWS Lambda function to aggregate the metrics.

Monitor and verify results

Run the following steps to monitor and verify the test results.

1. Run the following command to get the details of the execution. Replace executionArn with your state machine ARN.

   aws stepfunctions describe-execution --execution-arn <executionArn>

2. When the status shows SUCCEEDED, run the following commands to check the processed output from the LogAnalyticsSummaryTableName DynamoDB table. Replace the value LogAnalyticsSummaryTableName with the value from the sam deploy output.

   aws dynamodb scan --table-name <LogAnalyticsSummaryTableName>

3. Check that hourly ERROR, WARN, and INFO logs statistics are saved in the DynamoDB table. The following is a sample output:

   {
       "Items": [
           {
               "ProcessingTime": {
                   "S": "2025-10-07T23:45:10.790Z"
               },
               "WarningCount": {
                   "N": "2"
               },
               "HourOfDay": {
                   "S": "13"
               },
               "TotalRecords": {
                   "N": "5"
               },
               "ErrorCount": {
                   "N": "3"
               },
               "InfoCount": {
                   "N": "0"
               },
               "HourKey": {
                   "S": "2025-10-08 13"
               }
           },
           {
               "ProcessingTime": {
                   "S": "2025-10-07T23:45:07.456Z"
               },
               "WarningCount": {
                   "N": "3"
               },
               "HourOfDay": {
                   "S": "09"
               },
               "TotalRecords": {
                   "N": "6"
               },
               "ErrorCount": {
                   "N": "2"
               },
               "InfoCount": {
                   "N": "1"
               },
               "HourKey": {
                   "S": "2025-10-08 09"
               }
           },
           …
   ],
       "Count": 24,
       "ScannedCount": 24,
       "ConsumedCapacity": null
   }

4. Run the following command to check the output of the Step Functions state machine execution output.

   aws stepfunctions describe-execution --execution-arn <executionArn> --query 'output' --output text

   The following is a sample output:

   {
     "Summary": {
       "date": "2025-10-08",
       "totalErrors": 50,
       "totalWarnings": 41,
       "totalRecords": 133,
       "hourlyBreakdown": {
         "00": {
           "errors": 1,
           "warnings": 3,
           "records": 5
         },
         "01": {
           "errors": 1,
           "warnings": 1,
           "records": 5
         },
         "02": {
           "errors": 2,
           "warnings": 3,
           "records": 5
         },
         "03": {
           "errors": 3,
           "warnings": 2,
           "records": 7
         },
   …
   …
       "generatedAt": "2025-10-08T05:19:05.603889"
     }
   }

   The output of the Step Functions state machine shows the daily summary insights of the log files created by the Lambda function.

Clean up

To avoid costs, remove all resources created for this post once you’re done. Run the following command after replacing amzn-s3-demo-bucket1 with your own bucket name to delete the resources you deployed for this post’s solution:

aws s3 rm s3://amzn-s3-demo-bucket1 --recursive
sam delete
rm -rf logs/

Conclusion

In this post, you learned how AWS Step Functions Distributed Map can use prefix-based iteration with LOAD_AND_FLATTEN transformation to read and process multiple data objects from Amazon S3 buckets directly. You no longer need one step to process object metadata and another to load the data objects. Loading and flatting in one step is particularly valuable for data processing pipelines, batch operations, and event-driven architectures where objects are continually added to S3 locations. By eliminating the need to maintain object manifests, you can build more resilient, dynamic data processing workflows with less code and fewer moving parts.

New input sources for Distributed Map are available in all commercial AWS Regions where AWS Step Functions is available. To get started, you can use the Distributed Map mode today in the AWS Step Functions console. To learn more, visit the Step Functions developer guide.

For more serverless learning resources, visit Serverless Land.