标签: aws-step-functions

假设我的 Step Function 的一部分如下所示：

"ChoiceStateX": {
  "Type": "Choice",
  "Choices": [
    {
      "Variable": "$.value",
      "NumericEquals": 0,
      "Next": "ValueIsZero"
    }
  ],
  "Default": "DefaultState"
},
"ValueIsZero": {
  "Type" : "Task",
  "Resource": "arn:aws:lambda:******:function:Zero",
  "Next": "NextState"
},
"DefaultState": {
  "Type" : "Task",
  "Resource": "arn:aws:lambda:******:function:NotZero",
  "Next": "NextState"
}

假设该状态的输入是：

{
   "value": 0,
   "output1": object1,
   "output2": object2,
}

我的问题是我必须传递output1给ValueIsZerostateoutput2和DefaultState. 我知道可以改变和InputPath状态。但这种方式对我来说是不可接受的，因为我也从其他一些州打电话给这些州。ValueIsZeroDefaultState

我尝试ChoiceStateX像下面这样修改状态：

"ChoiceStateX": {
  "Type": "Choice",
  "Choices": [
    {
      "Variable": "$.value",
      "NumericEquals": 0,
      "OutputPath": …

有没有办法在 StepFunction 中创建选择规则来确定数组字段是否有任何元素？

当前的 StepFunction 文档没有列出任何特定于集合的比较运算符，所以我想知道是否可以在不实现单独的 lambda 来测试数组是否为空的情况下实现这一点？

将 Step Function 的部分输入发送到批处理作业的正确方法是什么？

我尝试使用Parameters.ContainerOverrides.Environment 设置和环境变量，如下所示：

"Parameters": {
    "ContainerOverrides": {
      "Environment": [
        {
          "Name": "PARAM_1",
          "Value": "$.param_1"
        }

步骤函数输入如下所示：

{
  "param_1": "value-goes-here"
}

但批处理作业最终会在 PARAM_1 环境变量中使用文字“$.param_1”进行调用。

我在 python 中有一个胶水工作，我从一个步骤函数调用它。步骤函数成功启动作业。作业成功完成。但步骤函数永远不会移动到下一步。步骤功能是否需要一些配置/权限来响应作业成功？python 脚本中需要做什么？

这是阶跃函数（状态机）定义：

"MyGlueTask": {
  "Type": "Task",
  "Resource": "arn:aws:states:::glue:startJobRun.sync",
  "Parameters": {
    "JobName": "my_glue_job"
  },
  "ResultPath": "$.MyGlueTask",
  "Next": "NextGlueJob"
}

我有一个步骤函数，它从许多并行步骤开始（每个并行步骤都是一个 lambda 调用），然后是一个完成一些最终处理的完成步骤。

它可以在这里可视化（下面也给出了状态函数定义的编辑版本）。我知道您可以在并行步骤周围添加 try/catch 逻辑，但如果我的理解是正确的，则不会阻止其他并行步骤继续并且不会将它们全部发送到不同的状态。

理想情况下，我想要的是，如果任何并行步骤因任何原因失败，所有当前步骤（以及未来的步骤）都将被取消，并且它们永远不会进入 Finalize 阶段，而是进入第三个状态（称之为 Error恢复）用于不同的执行。这个工作流程可行吗？如果是这样，是否可以保证在进入恢复状态之前所有并行步骤都已停止？

阶跃函数定义

{
  "Comment": "An example of the Amazon States Language using a map state to process elements of an array with a max concurrency of 2.",
  "StartAt": "Map",
  "States": {
    "Map": {
      "Type": "Map",
      "ItemsPath": "$.items",
      "Parameters": {
        ...
      },
      "MaxConcurrency": 2,
      "Next": "Finalize",
      "Iterator": {
        "StartAt": "Parallel Step",
        "States": {
          "Parallel Step": {
            "Type": "Task",
            "Resource": "arn:aws:states:::lambda:invoke",
            "Parameters": {
              "FunctionName": "arn:aws:lambda:us-east-1:<>:function:lambda-parallel-step:$LATEST",
              "Payload": {
                "Input.$": "$"
              }
            },
            "OutputPath": "$.Payload", …

我在账户 A 中有 Step Function，在账户 B 中有 lambda。但是在运行 step 函数时，它给出：

An error occurred while executing the state 'lambdaB' (entered at the event id #2). The resource belongs to a different account from the running execution.

有什么办法可以实现这种配置。

之前曾使用 AWS 步骤函数进行 lambda 编排。这一直运作良好。设置每个 lambda 的 result_path 会将参数传递给后续 lambda。

但是，我现在需要运行一个 Fargate 任务，然后将参数从该 Fargate 任务传递给后续的 lambda。我创建了一个 python 脚本，充当容器定义中的入口点。显然，在 lambda 函数中，handler(event, context)充当入口点，通过定义 a ，return {"return_object": "hello_world"}可以轻松地将长参数传递到状态机的下一个状态。

但就我而言，我有一个任务定义，其中包含从此 Dockerfile 创建的容器定义：

FROM python:3.7-slim

COPY my_script.py /my_script.py
RUN ln -s /python/my_script.py /usr/bin/my_script && \
chmod +x /python/my_script.py

ENTRYPOINT ["my_script"]

因此，我能够调用状态机并且它将按预期执行 my_script。但是如何从这个 python 脚本获取输出并将其传递到状态机中的另一个状态呢？

我找到了一些有关如何传递输入的文档，但没有找到传递输出的示例。

我正在尝试向我的步骤函数状态机添加一些错误处理，并且我在特定情况下遇到了问题，例如，如果我在某个步骤上失败，我有一个捕获将其发送到另一个函数，但是在该函数我需要使用最后一步输入数据（失败的数据）执行逻辑，但是，我刚刚收到一个错误，有没有办法传递带有错误的输入数据？

Send Notification:
        Type: Task
        Resource: !GetAtt CommunicationSendCertifiedEmail.Arn
        Retry:
          - ErrorEquals:
            - TierOneError
            IntervalSeconds: 2
            MaxAttempts: 9
            BackoffRate: 1.5
          - ErrorEquals:
            - TierTwoError
            IntervalSeconds: 4
            MaxAttempts: 6
            BackoffRate: 1.5
          - ErrorEquals:
            - TierThreeError
            IntervalSeconds: 4
            MaxAttempts: 3
            BackoffRate: 2
          - ErrorEquals:
            - States.Timeout
            - States.Runtime
            - States.TaskFailed
            - Lambda.ServiceException
            IntervalSeconds: 4
            MaxAttempts: 3
            BackoffRate: 2
        Catch:
          - ErrorEquals:
            - States.ALL
            Next: Send Email Notification

在“发送电子邮件通知”上，我需要在执行失败之前输入，但正如我之前所说，我只是得到这样的信息：

{
  "Error": "Error",
  "Cause": "{\"errorType\":\"Error\",\"errorMessage\":...
}

我想要得到这样的东西：

{
  "data": "{...}",
  "Error": "Error",
  "Cause": "{\"errorType\":\"Error\",\"errorMessage\":...
} …

我能够使用 lambda 函数执行将数据从源存储桶复制到目标存储桶的任务，但是，在 Step 函数中执行 lambda 函数时出现错误。以下是我从头开始遵循的步骤。

1. 选择的区域是 ap-south-1
2. 创建了 2 个桶。源存储桶：start.bucket & 目标存储桶：final.bucket
3. 使用以下信息创建 Lambda 函数：
   • 作者从零开始
   • 函数名称：复制复制
   • 运行时：Python 3.8
   • 已创建 lambda IAM 角色：LambdaCopy 并提供必要的策略（S3 完全访问权限和 Step 函数完全访问权限）并将其附加到该函数。
   • 添加触发器并选择：
     • S3
     • 桶：start.bucket
     • 事件类型：所有对象创建事件
   • 我在GeeksforGeeks中找到了一段python代码，并应用到了代码部分。

import json
import boto3
s3_client=boto3.client('s3')

# lambda function to copy file from 1 s3 to another s3
def lambda_handler(event, context):
    #specify source bucket
    source_bucket_name=event['Records'][0]['s3']['bucket']['name']
    #get object that has been uploaded
    file_name=event['Records'][0]['s3']['object']['key']
    #specify destination bucket
    destination_bucket_name='final.bucket'
    #specify from where file needs to be copied
    copy_object={'Bucket':source_bucket_name,'Key':file_name}
    #write copy …

我正在我的 ECS fargate 集群上创建并运行任务。

任务定义（带有角色）和 Fargate 集群已创建。

当我在步骤函数中使用运行任务步骤时，出现以下错误，

{
  "Error": "ECS.AccessDeniedException",
  "Cause": "User: arn:aws:sts::xxxxxxxxxx:assumed-role/StepFunctions-my-state-machine-role-xxxxxxxxxx/xxxxxxxxxx is not authorized to perform: iam:PassRole on resource: arn:aws:iam::xxxxxxxxxx:role/my-app-dev-exec because no identity-based policy allows the iam:PassRole action (Service: AmazonECS; Status Code: 400; Error Code: AccessDeniedException; Request ID: xxxxxxxxxx-xxxxxxxxxx-xxxxxxxxxx; Proxy: null)"
}

附加到步骤函数的角色具有以下策略（根据 AWS 提供的文档https://docs.aws.amazon.com/step-functions/latest/dg/ecs-iam.html）

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "ecs:RunTask"
            ],
            "Resource": [
                "arn:aws:ecs:eu-west-1:xxxxxxxxxx:task-definition/*:*"
            ]
        },
        {
            "Effect": "Allow",
            "Action": [
                "ecs:StopTask",
                "ecs:DescribeTasks"
            ],
            "Resource": [
                "arn:aws:ecs:eu-west-1:xxxxxxxxxx:task/*"
            ]
        }, …