【发布时间】：2019-01-09 08:20:48
【问题描述】：
【问题讨论】：
标签： amazon-web-services aws-step-functions
【发布时间】：2019-01-09 08:20:48
【问题描述】：
可能如下图所示
并行状态应该是这样的
"MyParallelState": {
"Type": "Parallel",
"InputPath": "$",
"OutputPath": "$",
"ResultPath": "$.ParallelResultPath",
"Next": "SetCartCompleteStatusState",
"Branches": [
{
"StartAt": "UpdateMonthlyUsageState",
"States": {
"UpdateMonthlyUsageState": {
"Type": "Task",
"InputPath": "$",
"OutputPath": "$",
"ResultPath": "$.UpdateMonthlyUsageResultPath",
"Resource": "LambdaARN",
"End": true
}
}
},
{
"StartAt": "QueueTaxInvoiceState",
"States": {
"QueueTaxInvoiceState": {
"Type": "Task",
"InputPath": "$",
"OutputPath": "$",
"ResultPath": "$.QueueTaxInvoiceResultPath",
"Resource": "LambdaARN",
"End": true
}
}
}
MyParallelState 的输出将以数组形式填充，来自Parallel state 中的每个状态。它们填充在ParallelResultPath 对象中，并将被传递到下一个状态
{
"ParallelResultPath": [
{
"UpdateMonthlyUsageResultPath": Some Output
},
{
"QueueTaxInvoiceResultPath": Some Output
}
]
}
【讨论】：
并行任务总是输出一个数组（每个分支包含一个条目）。
您可以告诉 AWS 步进函数将输出附加到原始输入中的新（或现有）属性中，并在并行状态定义中使用 "ResultPath": "$.ParallelOut"，但这似乎不是您想要实现的目标。
要合并并行任务的输出，您可以利用"Type": "Pass" 状态定义转换以应用于 JSON 文档。
例如，在下面的状态机中，我正在转换一个 JSON 数组...
[
{
"One": 1,
"Two": 2
},
{
"Foo": "Bar",
"Hello": "World"
}
]
...分成几个属性
{
"Hello": "World",
"One": 1,
"Foo": "Bar",
"Two": 2
}
{
"Comment": "How to convert an array into properties",
"StartAt": "warm-up",
"States": {
"warm-up": {
"Type": "Parallel",
"Next": "array-to-properties",
"Branches": [
{
"StartAt": "numbers",
"States": {
"numbers": {
"Type": "Pass",
"Result": {
"One": 1,
"Two" : 2
},
"End": true
}
}
},
{
"StartAt": "words",
"States": {
"words": {
"Type": "Pass",
"Result": {
"Foo": "Bar",
"Hello": "World"
},
"End": true
}
}
}
]
},
"array-to-properties": {
"Type": "Pass",
"Parameters": {
"One.$": "$[0].One",
"Two.$": "$[0].Two",
"Foo.$": "$[1].Foo",
"Hello.$": "$[1].Hello"
},
"End": true
}
}
}
【讨论】：
-
这对我有用。如果您使用 YAML 来描述 Serverless 框架上的状态机，请确保不要为每个参数使用“-”，因为这将为每个参数创建一个单独的对象。
-
是否保证
numbers步骤的输出仅出现在 0 索引处，结果数组中的输出也可以混洗，对吗？ -
同意@techytushar 是否有任何文件可以保证这一点？
-
@techytushar 是的，它确实维护了订单。见states-language.net/spec.html#parallel-state
-
TL;DR（来自上述文档）-
The elements of the output array correspond to the branches in the same order that they appear in the "Branches" array
您的图表在技术上是错误的，因为没有状态可以为其Next 任务设置多个状态。您不能通过提供多个状态名称以 StartAt 启动状态机。此外，即使有可能，我也看不出为什么要运行两个并行状态，而不是一个并行状态，所有子状态都会分成两个。
【讨论】：
我们可以使用 ResultSelector 和 Result Path 将结果组合成一个对象
我们有一个类似的并行状态：
{
"StartAt": "ParallelBranch",
"States": {
"ParallelBranch": {
"Type": "Parallel",
"ResultPath": "$",
"InputPath": "$",
"OutputPath": "$",
"ResultSelector": {
"UsersResult.$": "$[1].UsersUpload",
"CustomersResult.$": "$[0].customersDataUpload"
},
"Branches": [
{
"StartAt": "customersDataUpload",
"States": {
"customersDataUpload": {
"Type": "Pass",
"ResultPath": "$.customersDataUpload.Output",
"Result": {
"CompletionStatus": "success",
"CompletionDetails": null
},
"Next": "Wait2"
},
"Wait2": {
"Comment": "A Wait state delays the state machine from continuing for a specified time.",
"Type": "Wait",
"Seconds": 2,
"End": true
}
}
},
{
"StartAt": "UsersUpload",
"States": {
"UsersUpload": {
"Type": "Pass",
"Result": {
"CompletionStatus": "success",
"CompletionDetails": null
},
"ResultPath": "$.UsersUpload.Output",
"Next": "Wait1"
},
"Wait1": {
"Comment": "A Wait state delays the state machine from continuing for a specified time.",
"Type": "Wait",
"Seconds": 1,
"End": true
}
}
}
],
"End": true
}
},
"TimeoutSeconds": 129600,
"Version": "1.0"
}
输出会是这样的：
{
"UsersResult": {
"Output": {
"CompletionStatus": "success",
"CompletionDetails": null
}
},
"CustomersResult": {
"Output": {
"CompletionStatus": "success",
"CompletionDetails": null
}
}
}
【讨论】：
-
如果假设一个分支作业失败，它会等待另一个作业，还是分支会立即失败
-
默认情况下，如果一个分支作业失败，所有分支都会立即失败。
这对我有用
"Transform And Freeze": {
"Type": "Parallel",
"InputPath": "$",
"Branches": [
{
"StartAt": "Transform Status",
"States": {
"Transform Status": {
"Type": "Map",
"ItemsPath": "$",
"MaxConcurrency": 25,
"Iterator": {
"StartAt": "Transform",
"States": {
"Transform": {
"Type": "Task",
"Resource": "${TransformFunction}",
"End": true
}
}
},
"End": true
}
}
},
{
"StartAt": "Freeze Status",
"States": {
"Freeze Status": {
"Type": "Map",
"MaxConcurrency": 25,
"Iterator": {
"StartAt": "Freeze",
"States": {
"Freeze Transactions": {
"Type": "Task",
"Resource": "${FreezeFunction}",
"End": true
}
}
},
"End": true
}
}
}
],
"ResultPath" : "$.parts",
"Next": "SetParallelOutput",
"Catch": [
{
"ErrorEquals": [
"States.ALL"
],
"ResultPath": "$.exception",
"Next": "Error Handler"
}
]
},
"SetParallelOutput": {
"Type": "Pass",
"Parameters": {
"foo.$": "$.foo",
"bar.$": "$.bar",
"parts.$": "$.parts[0]"
},
"Next": "Target Type"
},
【讨论】：