如何在AWS Step Functions中过滤数据并按状态存储至S3？

你不需要必须依赖Lambda来完成过滤和写入，Step Functions本身支持直接处理数组过滤，同时可以直接调用S3 API完成文件存储；当然如果逻辑复杂，Lambda也是可选方案。

方案一：仅用Step Functions内置功能完成（无Lambda） #

Step Functions的内置数组处理函数（比如States.ArrayPartition、States.ArrayFilter）可以直接在状态中完成数据拆分，再通过AWS SDK集成的Task状态直接调用S3的PutObject接口写入文件，全程无需Lambda。

具体实现步骤 #

1. 拆分过滤数组：使用Pass状态，通过内置函数将原数组按Status拆分为成功、失败两个子数组，并将结果存入上下文：

{
  "Type": "Pass",
  "Parameters": {
    "successItems.$": "States.ArrayPartition($, $.Status == 'Success')[0]",
    "failedItems.$": "States.ArrayPartition($, $.Status == 'Success')[1]"
  },
  "ResultPath": "$.filtered",
  "Next": "WriteSuccessToS3"
}

States.ArrayPartition会返回包含两个数组的结果，第一个是符合条件的Success项，第二个是Failed项。

2. 写入S3文件：用Task状态直接调用S3接口，分别写入两个文件：

{
  "Type": "Task",
  "Resource": "arn:aws:states:::s3:putObject",
  "Parameters": {
    "Bucket": "你的存储桶名称",
    "Key": "success-items.json",
    "Body.$": "States.Stringify($.filtered.successItems)"
  },
  "Next": "WriteFailedToS3"
}

同理创建WriteFailedToS3状态，将failedItems写入failed-items.json即可。

方案二：用Lambda处理复杂过滤（可选） #

如果你的过滤逻辑不止是简单匹配Status字段（比如多条件判断、数据格式转换等），可以用Lambda先完成过滤，再在同一个Lambda里写入S3，或者返回过滤结果给Step Functions后再调用S3写入。

Lambda核心逻辑示例：

import boto3
import json

s3 = boto3.client('s3')

def lambda_handler(event, context):
    # 过滤数据
    success_items = [item for item in event if item['Status'] == 'Success']
    failed_items = [item for item in event if item['Status'] == 'Failed']
    
    # 写入S3
    s3.put_object(
        Bucket='你的存储桶名称',
        Key='success-items.json',
        Body=json.dumps(success_items)
    )
    s3.put_object(
        Bucket='你的存储桶名称',
        Key='failed-items.json',
        Body=json.dumps(failed_items)
    )
    
    return {
        'successCount': len(success_items),
        'failedCount': len(failed_items)
    }

之后在Step Functions中用一个Task状态调用该Lambda即可。

选择建议 #

• 仅需简单字段过滤时，优先用方案一，减少Lambda的维护成本和冷启动开销；
• 过滤逻辑复杂或需额外数据处理时，方案二更灵活。

内容的提问来源于stack exchange，提问作者PBandJ