AWS Step Function简介

 产生StepFunction可以用console的template产生，节省时间

如何在Step function用变量

每一个task都有input和output，如果一个task的TaskStateEntered(AWS Console)的input是

{"input": {

       "catalogId": "abc-cde"

    } 

}

State machine中，这样取值$.catalogId, 根目录对应的是input. 这个task也并不需要写ResultPath(只做filter之用).

input对应$.

output对应$.output.

content对应$..

通过选择，可以选取Lambda InputStream的输入，如

Parameters: { 

    "FunctionName": "xxx",

    "Payload": {

         "execution.$": "$$.Execution.Id",

         "categoryId.$":: "$.catalogId",

         "runId.$": "$output.JobRunState"

    }

}

Payload对应Lambda的InputStream的输入

这个task(假设是lambda)是按照以下方式写入output:

public void handle(InputStream in, OutputStream out, Context context) {

     HashMap map;

     map.put("catalogId, "abc-cde");

    String json = objectwriter.toJson(map);

    out.write(json);

}

Map换成一个object也是一样的。

Context变量

Context含有step function execution arn，用的时候用$$.

"Execution.$": "$$.Execution.Id"

StepFunction可以invoke Glue

Step function能支持的Lambda的参数包括Payload，这是lambda的输入

"GetJobParams": {

   "Type": "Task",

   "Resource": "arn:aws:states:::lambda:invoke",

   "Parameters": {

         "FunctionName": "arn...Lambda",

         "Payload": {

                 "execution.$": "$$.Execution.Id",

                 "catalogId.$": "$.catalogId"

          }

     }

}

$.catalogId是原有的lambda input的一个attribute，这里是将execution添加到Lambda的input中。

StepFunction可以invoke Glue

"Glue StartJobRun": {

      "Type": "Task",

      "Resource": "arn:aws:states:::glue:startJobRun.sync",

      "Parameters": {

        "JobName": "my-etl-job"

      },

      "Next": "ValidateOutput"

    },

Resource不能改，一般来说是用ARN，但Glue没有ARN，Step function通过JobName来定位Glue job。JobRunId不能用自定义格式或者不能加入作为参数，否则会说resource找不到。返回值是Id (JobRunId)，JobRunState，ErrorMessage, StartedOn等等。

如果某个Task failed还想继续执行下一个任务，可以用Catch，如ETL job失败，还是想将失败状态写入数据库

"Glue StartJobRun": {

      "Type": "Task",

      "Resource": "arn:aws:states:::glue:startJobRun.sync",

      "Parameters": {

        "JobName": "my-etl-job"

      },

      "Catch": [ {

            "ErrorEquals": ["States.Timeout", "States.TaskFailed", "HandledError"],

            "Next": "ValidateOutput"

         } ],

      "ResultPath": "$.output",

      "Next": "ValidateOutput"

    },

$.output存储Glue的输出结果，下一个task如lambda可以使用。

如何给一个task赋值

刚才讲到怎么调用变量，现在讲怎么给一个task的参数赋值。关键在于尾部加入.$

"Glue StartJobRun": {

        "JobName": "my-etl-job"，

        "Arguments": {

               "--catalog_id.$": "$.catalogId"

          }

      },

Step function只支持这些参数

Lambda retry

lambda本身的retry并不支持，需要再step function里面定义retry

"Retry": [

        {

          "ErrorEquals": [

            "States.ALL"

          ],

          "IntervalSeconds": 30,

          "MaxAttempts": 2,

          "BackoffRate": 2

        }

      ],