如何在AWS Step Functions中遍历数组并附加额外数据？

我正尝试处理DynamoDB查询结果，比如在Map组件中处理$.DDBResult，数据结构如下：

{
  ...,
  "LookBackSeconds":"60",                     // 基于请求的变量
  "OneMinuteAgo":"2023-08-14T15:59:00.000Z", // 通过Lambda函数用$.LookBackSeconds计算得出
  "DDBResult": {
    "Count": 3,
    "Items": [
      {
        "Status": {
          "S": "ONLINE"
        },
        "Location": {
          "S": "North"
        },
        "LastValidationDateTime": {
          "S": "2023-08-14T14:55:55.144Z"
        }
      },
      {
        "Status": {
          "S": "ONLINE"
        },
        "Location": {
          "S": "South"
        },
        "LastValidationDateTime": {
          "S": "2023-08-14T14:55:54.890Z"
        }
      },
      {
        "Status": {
          "S": "ONLINE"
        },
        "Location": {
          "S": "East"
        },
        "LastValidationDateTime": {
          "S": "2023-08-14T14:55:52.968Z"
        }
      }
    ],
    "ScannedCount": 3
  }
}

我知道可以把$.DDBResult.Items设为ItemsPath，但问题是在Map里需要用到额外的$.OneMinuteAgo数据项。本来想遍历$.DDBResult.Items，基于原数据创建新数组并把$.OneMinuteAgo加进去，但找不到遍历数组构建新数组的方法，求解决办法。

方法1：直接在Map里引用外部变量 #

其实不用提前修改数组，Step Functions的Map组件支持在迭代时直接引用状态机上下文里的外部变量。

只要把Map的ItemsPath设为$.DDBResult.Items，然后在ItemProcessor的任务参数里，直接引用$.OneMinuteAgo就行：

"Parameters": {
  "CurrentItem.$": "$",
  "OneMinuteAgo.$": "$.OneMinuteAgo"
}

这样每个迭代任务都能拿到当前Item和OneMinuteAgo的值，完全不用改原数组结构。

方法2：用Transform状态预处理数组 #

如果一定要把OneMinuteAgo嵌入到每个Item里，可以用Step Functions的Transform状态（需要ASL v1.0以上版本），用内置的ArrayMap和Merge操作生成新数组：

"TransformItems": {
  "Type": "Transform",
  "Parameters": {
    "ProcessedItems.$": "States.ArrayMap($.DDBResult.Items, (item) => States.Merge(item, {OneMinuteAgo: $.OneMinuteAgo}))"
  },
  "Next": "YourMapState"
}

这个Transform会遍历原Items数组，把每个Item和OneMinuteAgo合并，生成新的ProcessedItems数组。之后把Map的ItemsPath设为$.ProcessedItems就行。

方法3：用Lambda预处理数组 #

要是上面的内置方法用不了，写个简单的Lambda函数处理也很方便：

def lambda_handler(event, context):
    items = event['DDBResult']['Items']
    one_minute_ago = event['OneMinuteAgo']
    # 把OneMinuteAgo按DynamoDB格式添加到每个Item里
    new_items = [item | {'OneMinuteAgo': {'S': one_minute_ago}} for item in items]
    # 返回包含新数组的结果
    return {
        **event,
        'ProcessedItems': new_items
    }

调用这个Lambda后，Map的ItemsPath设为$.ProcessedItems就可以迭代处理带OneMinuteAgo的数组了。

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