詳解 AWS Step Functions: プログラム言語のように記法を理解しよう

{
  "Comment": "A description of my state machine",
  "StartAt": "現在のUNIX時刻を取得",
  "States": {
    "現在のUNIX時刻を取得": {
      "Type": "Pass",
      "Next": "UNIX時刻をJSTでフォーマット",
      "Output": {
        "TimeStamp": "{% $millis() %}"
      }
    },
    "UNIX時刻をJSTでフォーマット": {
      "Type": "Pass",
      "End": true,
      "Output": {
        "JstTimeStamp": "{% $fromMillis($states.input.TimeStamp, '[M01]/[D01]/[Y0001] [h#1]:[m01][P]', '+0900') %}"
      }
    }
  },
  "QueryLanguage": "JSONata"
}

{
  "StartAt": "変数宣言",
  "States": {
    "変数宣言": {
      "Type": "Pass",
      "Assign": {
        "MeesageTemplate": "{% \"今日のおやつは「\" & $states.context.Execution.Input.Oyatsu & \"」です。\" %}"
      },
      "Next": "変数展開"
    },
    "変数展開": {
      "Type": "Pass",
      "End": true,
      "Output": {
        "Output": "{% $MeesageTemplate %}"
      }
    }
  },
  "QueryLanguage": "JSONata"
}

{
  "Comment": "A description of my state machine",
  "StartAt": "variable変数を設定",
  "States": {
    "variable変数を設定": {
      "Type": "Pass",
      "Next": "Choice",
      "Assign": {
        "variable": "test"
      }
    },
    "Choice": {
      "Type": "Choice",
      "Choices": [
        {
          "Next": "trueの場合の操作",
          "Condition": "{% $exists($variable) %}",
          "Comment": "Trueの場合"
        }
      ],
      "Default": "falseの場合の操作"
    },
    "trueの場合の操作": {
      "Type": "Pass",
      "End": true,
      "Output": {
        "SetVariable": true
      }
    },
    "falseの場合の操作": {
      "Type": "Pass",
      "End": true,
      "Output": {
        "SetVariable": false
      }
    }
  },
  "QueryLanguage": "JSONata"
}

ループ

{
  "Comment": "A description of my state machine",
  "StartAt": "カウンタを設定",
  "States": {
    "カウンタを設定": {
      "Type": "Pass",
      "Next": "Choice",
      "Output": {
        "Counter": 0
      },
      "Assign": {
        "Counter": 0
      }
    },
    "Choice": {
      "Type": "Choice",
      "Choices": [
        {
          "Condition": "{% $Counter <= 3 %}",
          "Comment": "Trueの場合",
          "Next": "ループ中の処理"
        }
      ],
      "Default": "ループ終了"
    },
    "ループ終了": {
      "Type": "Pass",
      "End": true
    },
    "ループ中の処理": {
      "Type": "Pass",
      "Next": "カウンタを加算"
    },
    "カウンタを加算": {
      "Type": "Pass",
      "Output": {
        "SetVariable": true
      },
      "Assign": {
        "Counter": "{% $Counter + 1 %}"
      },
      "Next": "Choice"
    }
  },
  "QueryLanguage": "JSONata"
}

イテレーション

{
  "StartAt": "要素を設定",
  "States": {
    "要素を設定": {
      "Type": "Pass",
      "Output": {
        "Counter": 0
      },
      "Assign": {
        "Items": [
          "りんご",
          "ばなな",
          "ぶどう"
        ]
      },
      "Next": "Map"
    },
    "Map": {
      "Type": "Map",
      "ItemProcessor": {
        "ProcessorConfig": {
          "Mode": "INLINE"
        },
        "StartAt": "メッセージを出力",
        "States": {
          "メッセージを出力": {
            "Type": "Pass",
            "End": true,
            "Output": {
              "Message": "{% \"フルーツは「\" & $states.input & \"」です。\" %}"
            }
          }
        }
      },
      "Items": "{% $Items %}",
      "MaxConcurrency": 1,
      "Next": "各要素のメッセージを取得"
    },
    "各要素のメッセージを取得": {
      "Type": "Pass",
      "End": true,
      "Output": {
        "Messages": "{% $states.input.Message %}"
      }
    }
  },
  "QueryLanguage": "JSONata"
}

ユーザー関数として呼び出されるステートマシン(子)

{
  "Comment": "A description of my state machine",
  "StartAt": "実行入力からメッセージを取得",
  "States": {
    "実行入力からメッセージを取得": {
      "Type": "Pass",
      "End": true,
      "Output": {
        "Message": "{% '渡されたメッセージ: ' & $states.context.Execution.Input.Message %}"
      }
    }
  },
  "QueryLanguage": "JSONata"
}

ユーザー関数を同期処理で呼び出すステートマシン(親)

{
  "StartAt": "Step Functions StartExecution",
  "States": {
    "Step Functions StartExecution": {
      "Type": "Task",
      "Resource": "arn:aws:states:::states:startExecution.sync:2",
      "Arguments": {
        "StateMachineArn": "<子のステートマシンのARN>",
        "Input": {
          "Message": "Hello From Parent."
        }
      },
      "End": true
    }
  },
  "QueryLanguage": "JSONata"
}

ポーリング処理をして処理完了を待つステートマシン

{
  "StartAt": "クエリ実行",
  "States": {
    "クエリ実行": {
      "Type": "Task",
      "Arguments": {
        "Database": "redshift-databasename}",
        "ClusterIdentifier": "redshift-cluster-identifier",
        "Sql": "sql"
      },
      "Resource": "arn:aws:states:::aws-sdk:redshiftdata:executeStatement",
      "Next": "待機(10秒)",
      "Output": {
        "Id": "{% $states.result.Id %}"
      }
    },
    "待機(10秒)": {
      "Type": "Wait",
      "Seconds": 10,
      "Next": "ポーリング処理",
      "Output": {
        "Id": "{% $states.input.Id %}"
      }
    },
    "ポーリング処理": {
      "Type": "Task",
      "Arguments": {
        "Id": "{% $states.input.Id %}"
      },
      "Resource": "arn:aws:states:::aws-sdk:redshiftdata:describeStatement",
      "Next": "ポーリング結果判定",
      "Output": {
        "Id": "{% $states.input.Id %}",
        "Status": "{% $states.result.Status %}",
        "HasResultSet": "{% $states.result.HasResultSet %}",
        "ResultRows": "{% $states.result.ResultRows %}",
        "ResultSize": "{% $states.result.ResultSize %}"
      }
    },
    "ポーリング結果判定": {
      "Type": "Choice",
      "Choices": [
        {
          "Next": "待機(10秒)",
          "Condition": "{% $states.input.Status="SUBMITTED" or $states.input.Status="STARTED" or $states.input.Status="PICKED" %}",
          "Output": {
            "Id": "{% $states.input.Id %}",
            "Status": "{% $states.input.Status %}"
          }
        },
        {
          "Next": "成功",
          "Condition": "{% $states.input.Status="FINISHED"%}",
          "Output": {
            "Id": "{% $states.input.Id %}",
            "Status": "{% $states.input.Status %}",
            "HasResultSet": "{% $states.input.HasResultSet %}",
            "ResultRows": "{% $states.input.ResultRows %}",
            "ResultSize": "{% $states.input.ResultSize %}"
          }
        }
      ],
      "Default": "UnexpectedStatusError",
      "Output": {
        "Id": "{% $states.input.Id %}",
        "Status": "{% $states.input.Status %}",
        "Error": "",
        "HasResultSet": "{% $states.input.HasResultSet %}",
        "ResultRows": "{% $states.input.ResultRows %}",
        "ResultSize": "{% $states.input.ResultSize %}"
      }
    },
    "UnexpectedStatusError": {
      "Type": "Fail",
      "Error": "UnexpectedStatusError",
      "Cause": "{% 'the commands exited unexpected status:' & $states.input.Status %}"
    },
    "成功": {
      "Type": "Succeed",
      "Output": {
        "Id": "{% $states.input.Id %}",
        "Status": "{% $states.input.Status %}",
        "Error": "",
        "HasResultSet": "{% $states.input.HasResultSet %}",
        "ResultRows": "{% $states.input.ResultRows %}",
        "ResultSize": "{% $states.input.ResultSize %}"
      }
    }
  },
  "QueryLanguage": "JSONata"
}

Map

{
  "StartAt": "要素を設定",
  "States": {
    "要素を設定": {
      "Type": "Pass",
      "Assign": {
        "Items": [
          "りんご",
          "ぶどう",
          "ばなな"
        ]
      },
      "Next": "Map"
    },
    "Map": {
      "Type": "Map",
      "ItemProcessor": {
        "ProcessorConfig": {
          "Mode": "INLINE"
        },
        "StartAt": "並列処理",
        "States": {
          "並列処理": {
            "Type": "Pass",
            "Output": {
              "Message": "{% $states.input %}"
            },
            "End": true
          }
        }
      },
      "Next": "並列処理のメッセージを取得",
      "Items": "{% $Items %}"
    },
    "並列処理のメッセージを取得": {
      "Type": "Pass",
      "End": true,
      "Output": {
        "Messages": "{% $states.input.Message %}"
      }
    }
  },
  "QueryLanguage": "JSONata"
}

Parallel

{
  "StartAt": "要素を設定",
  "States": {
    "要素を設定": {
      "Type": "Pass",
      "Assign": {
        "Item": "りんご"
      },
      "Next": "Parallel"
    },
    "Parallel": {
      "Type": "Parallel",
      "Next": "並列処理のメッセージを取得",
      "Branches": [
        {
          "StartAt": "並列処理(1)",
          "States": {
            "並列処理(1)": {
              "Type": "Pass",
              "Output": {
                "Message": "{% $Item %}"
              },
              "End": true
            }
          }
        },
        {
          "StartAt": "並列処理(2)",
          "States": {
            "並列処理(2)": {
              "Type": "Pass",
              "Output": {
                "Message": "{% $Item %}"
              },
              "End": true
            }
          }
        },
        {
          "StartAt": "並列処理(3)",
          "States": {
            "並列処理(3)": {
              "Type": "Pass",
              "Output": {
                "Message": "{% $Item %}"
              },
              "Assign": {
                "hoge": "hoge"
              },
              "End": true
            }
          }
        }
      ]
    },
    "並列処理のメッセージを取得": {
      "Type": "Pass",
      "End": true,
      "Output": {
        "Messages": "{% $states.input.Message %}"
      }
    }
  },
  "QueryLanguage": "JSONata"
}

再試行

{
  "StartAt": "Step Functions StartExecution",
  "States": {
    "Step Functions StartExecution": {
      "Type": "Task",
      "Resource": "arn:aws:states:::states:startExecution.sync:2",
      "Arguments": {
        "StateMachineArn": "arn:aws:states:REGION:ACCOUNT_ID:stateMachine:STATE_MACHINE_NAME",
        "Input": {
          "StatePayload": "Hello from Step Functions!",
          "AWS_STEP_FUNCTIONS_STARTED_BY_EXECUTION_ID": "error_test"
        }
      },
      "End": true,
      "Retry": [
        {
          "ErrorEquals": [
            "States.ALL"
          ],
          "BackoffRate": 2,
          "IntervalSeconds": 1,
          "MaxAttempts": 3
        }
      ]
    }
  },
  "QueryLanguage": "JSONata"
}

エラーキャッチ

{
  "StartAt": "Step Functions StartExecution",
  "States": {
    "Step Functions StartExecution": {
      "Type": "Task",
      "Resource": "arn:aws:states:::states:startExecution.sync:2",
      "Arguments": {
        "StateMachineArn": "arn:aws:states:REGION:ACCOUNT_ID:stateMachine:STATE_MACHINE_NAME",
        "Input": {
          "StatePayload": "Hello from Step Functions!",
          "AWS_STEP_FUNCTIONS_STARTED_BY_EXECUTION_ID": "error_test"
        }
      },
      "End": true,
      "Catch": [
        {
          "ErrorEquals": [
            "States.ALL"
          ],
          "Next": "エラー発生時の処理を定義"
        }
      ]
    },
    "エラー発生時の処理を定義": {
      "Type": "Pass",
      "End": true
    }
  },
  "QueryLanguage": "JSONata"
}