在 Step Function 中合并并行状态的 JSON 输出答案

【问题标题】：Merging JSON outputs of parallel states in Step Function在 Step Function 中合并并行状态的 JSON 输出
【发布时间】：2019-11-08 23:03:22
【问题描述】：

我在 Step Function 中有一个并行块。并行块的输入看起来像

{ "a": null, "b": null }

2 个并行状态是 populateA 和 populateB。 populateA 的输出是

{ "a": "A", "b": null }

populateB 的输出是

{ "a": null, "b": "B" }

并行块的输出是一个数组

[ { "a": "A", "b": null }, { "a": null, "b": "B" } ]

如何将两者合并成

{ "a": "A", "b": "B" }

【问题讨论】：

标签： amazon-web-services aws-step-functions

【解决方案1】：

如果 PopulateA 始终只填充 a 的值，而 PopulateB 只填充 b 的值，您可以在 Parallel State 之后使用带有参数的 Pass 状态（而不是 Result 字段）来组合输入。参数允许您通过使用 JSONPath 指定键值对来转换输入。

这是一个状态机示例：

{
"StartAt": "Parallel",
"States": {
    "Parallel": {
        "Type": "Parallel",
        "ResultPath": "$.CombinedOutput",
        "Next": "MergeOutputs",
        "Branches": [{
                "StartAt": "populateA",
                "States": {
                    "populateA": {
                        "Type": "Pass",
                        "Result": {
                            "a": "A",
                            "b": null
                        },
                        "End": true
                    }
                }
            },
            {
                "StartAt": "populateB",
                "States": {
                    "populateB": {
                        "Type": "Pass",
                        "Result": {
                            "a": null,
                            "b": "B"
                        },
                        "End": true
                    }
                }
            }
        ]
    },
    "MergeOutputs": {
        "Type": "Pass",
        "Parameters": {
            "a.$": "$.CombinedOutput[0].a",
            "b.$": "$.CombinedOutput[1].b"
        },
        "Next": "EndState"
    },
    "EndState": {
        "Type": "Pass",
        "End": true
    }
  }
}

【讨论】：

这对我有用。如果您使用 YAML 来描述 Serverless 框架上的状态机，请确保不要为每个参数使用“-”，因为这将为每个参数创建一个单独的对象。
请注意，如果您有稍微不同的情况，即 PopulateA 不仅填充 a 还创建具有其他字段的对象，并且 PopulateB 更新该对象上的字段，那么请注意这里的人，那么您不能使用此解决方案使其工作，但您可以编写一个简单的 lambda 来合并两个输出。
您能否提供一个输入和输出示例（针对 PopulateA 和 populateB）以及您预期的最终输出？

【解决方案2】：

恐怕在没有额外状态的情况下（目前）不可能做到这一点。您可以在 Parallel 状态之后添加额外的 Task 来合并数组元素。

附加任务的代码如下所示（Python）：

def merge_dicts(a, b):
    result = a.copy()
    for key, value in b.items():
        if value is not None:
            result[key] = value
    return result

def handler(event, context):
    result = {}
    for item in event:
        result = merge_dicts(result, item)
    return result

handler 从 Parallel 状态获取输出并将其合并到单个字典中。状态应该直接在 Parallel 之后。

请记住，此代码仅处理 flat-dict（不是嵌套的），因此它适用于简单的情况，但不适用于复杂的情况（我在此处添加它只是为了展示总体思路）。

【讨论】：

因此，您正在使用 python 代码控制状态机的步骤。你怎么知道状态机在哪一步？
@sphoenix 在这种情况下没关系，因为逻辑总是相同的。状态机创建者必须使用特定的 Lambda ARN 简单地创建任务状态继并行状态