标签: aws-step-functions

我有 AWS step machine，其中一个 step 用于使用 SNS 服务通知失败。我想从inputjson 中选择一些元数据到传出消息中。所以我试图将常量字符串与 jsonpath 连接起来，如下所示

"Notify Failure": {
      "Type": "Task",
      "Resource": "arn:aws:states:::sns:publish",
      "Parameters": {
        "Message.$": "A job submitted through Step Functions failed for document id $.document_id",
        "Subject":"Job failed",
        "TopicArn": "arn:aws:sns:us-west-2:xxxxxxx:xxxxxxxx"
      },
      "End": true
    }

document_id输入 json 中的属性之一在哪里

但是，当我尝试保存状态机定义时，出现错误

您的 ASL 定义有问题，请检查并重试 'Message.$' 字段的值必须是有效的 JSONPath

我正在开发一个项目，该项目从 AWS S3 获取一组输入数据，对其进行预处理和分割，启动 10K 批处理容器以在 AWS Batch 上并行处理分割数据，对数据进行后聚合并推送它到 S3。

我已经从其他项目中获得了 Airflow + Batch 的软件模式，但还没有处理 10k 并行任务的缩放因子。Airflow 很好，因为我可以查看哪些任务失败并在调试后重试任务。但是在一个 Airflow EC2 实例上处理这么多任务似乎是一个障碍。另一种选择是让一项任务启动 10k 容器并从那里监控它。

我没有使用 Step Functions 的经验，但听说它是 AWS 的 Airflow。Step Functions + Batch 在线看起来有很多模式。Step Functions 似乎是检查我的用例的好方法吗？您对失败的工作/重试任务的能力是否与使用 Airflow 获得相同的见解？

我有一个步进函数（如下图所示），当我突然遇到一个相当意外的结果时，我正在解决剩下的几个问题：

• 已完成的所有任务，成功完成无错误
• 所有剩余的任务都模拟取消（也没有错误）

我有点傻眼了。有没有人看到这个？有谁知道我将如何解决它？

我在 SAM/CloudFormation 模板中创建了一个 Step Function（父），它调用另一个 Step Function（子）。我正在按照使用服务集成模式从 Parent 调用 Child 的说明进行操作。但是我遇到了一个与 IAM 相关的（我认为）错误，我在通过 CLI 部署时无法解决。（错误显示在 CLI 输出中，因此它从未真正进入 AWS。之前有很多部署，因此changeset只是尝试使用此部署修改 Step Function。）

'arn:aws:iam::{Account-Number}:role/{Parent-Step-Function-Role-Name}' is not authorized to create managed-rule. (Service: AWSStepFunctions; Status Code: 400; Error Code: AccessDeniedException; Request ID: {Long-Id-Number})

为了获得我想要的同步行为（父调用子，等待子的执行完成，然后移动到下一个状态）我使用建议（来自上面的服务集成模式链接）创建一个任务（在我的 SAM 模板中）如下所示：

...More States...

"Call Child State": {
  "Type": "Task",
  "Next": "The Next State",
  "Resource": "arn:aws:states:::states:startExecution.sync",
  "Parameters": {  
    "Input": {
      "comment": "Hello World!"
    },
    "StateMachineArn": "${ChildStepFunction}",
    "Name": "ChildExecutionFromParent"
  }
},

...More States...

我已经为 Parent 定义了 IAM …

我有一个用例,我有一个AWS Step功能,当文件上传到S3时触发,从那里第一步运行ffprobe从外部服务获取文件的持续时间,如transloadit,其中写入输出回到S3.

我可以从该事件创建一个新的step函数,但是如果可以在原始step函数中有一个Await promise然后继续到下一个函数,我就会徘徊 - 考虑到ffprobe可能需要更长的时间才能恢复.

任何建议都非常感谢如何解决这个问题.

我想为传递给步进函数的参数设置一个默认值

例如，

"Parameters": {
   "foo.$": "$.foo" OR "bar" if "$.foo" not specified
}

有没有办法用 JSONPath 本地做到这一点，还是我必须使用选择+传递状态？

如果在输入中未指定参数时有一种方法不会中断，我什至会选择使用选择/通过。

如果我不包含"foo": ""在输入中，我会收到类似的错误"JSONPath ... could not be found in the input."

我有一个启动 lambda 的简单步骤函数，我正在寻找一种将参数（事件/上下文）传递给几个后续任务中的每一个的方法。我的步骤函数如下所示：

{
  "Comment": "A Hello World example of the Amazon States Language using an AWS Lambda function",
  "StartAt": "HelloWorld",
  "States": {
    "HelloWorld": {
      "Type": "Task",
      "Parameters": {
        "TableName": "table_example"
      },
      "Resource": "arn:aws:lambda:ap-southeast-2:XXXXXXX:function:fields_sync",
      "End": true
    }
  }
}

在用 Python 编写的 lambda 中，我使用了一个简单的处理程序，它是：

def lambda_handler(event, context):
    #...

事件和上下文如下所示（检查日志）：

开始请求 ID：f58140b8-9f04-47d7-9285-510b0357b4c2 版本：$LATEST

我找不到将参数传递给此 lambda 并在脚本中使用它们的方法。本质上，我想要做的是运行相同的 lambda，将几个不同的值作为参数传递。

有人可以指出我正确的方向吗？

我按照教程创建和调用步骤函数

我在api的GET请求中得到了输出

 {
  "executionArn": "arn:aws:states:ap-northeast-1:123456789012:execution:HelloWorld:MyExecution",
  "startDate": 1.486772644911E9
}

但是,而不是上面的响应,我想要我的步骤函数输出,由下面的结束状态给出.

{
   "name":"Hellow World"
}

怎么做到这一点？

我有一个程序执行几千蒙特卡罗模拟来预测结果; 我不能说他们真正预测的是什么,所以我将使用"无可争议的圣诞老人存在"中的另一个例子,因为这些算法的内容与问题无关.我想知道Monopoly董事会上每个广场的访问频率(预测哪些是最好的购买物业).为此,我模拟了数千个游戏并整理结果.我目前的实现是一个独立的C#应用​​程序,但我想将它移动到云端,以便我可以将其作为服务提供 - 每个用户都可以通过提交每个骰子的边数来获得个性化结果.

当前的实现也非常慢 - 它非常简单,因为每个模拟都是完全独立的,但我只有8个内核,因此在我的本地机器上完成大约50000个单独模拟的完整预测需要20分钟.

计划是让AWS lambda函数运行一个(或几个)模拟然后整理 - 基本上mapreduce它.我查看了使用AWS EMR(弹性MapReduce),但这对于我想要的太大了,旋转实例单独运行计算似乎比单独的整个计算花费更长时间(这对于多个小时离线分析,但我希望通过Web请求进行低延迟响应).

我认为理想的是:

Lambda 0 - 触发许多其他lambda函数,每个函数执行一小部分计算.Lambda 1..N - 并行进行多次模拟(数字不是常数).Lambda N + 1 - 整理所有结果并返回答案.

这里有一个lambda mapreduce框架:

https://github.com/awslabs/lambda-refarch-mapreduce

但它似乎有一个主要缺点 - 每次地图阶段完成时,它会将结果写入S3(我可以将其作为临时使用)然后通过事件触发新的lambda.触发lambda看看是否所有结果都已写入存储.如果不是,则结束,如果是,则执行还原步骤.这似乎是一个公平的解决方案,但我只是稍微关注a)当两个结果汇总在一起时,两个减速器都能计算结果？并且b)看起来好像它已经解雇了很多只是决定不运行的lambdas(我知道它们运行起来很便宜,但是每次模拟的数量加倍到两次 - 计算并且可能会减少 - 显然会使成本增加一倍).有没有办法在100个文件写入文件夹而不是每个文件后写入S3结果？

我看了一下使用步骤函数,但是我不确定如何在一步中并行激发多个lambdas并让它们在状态机转换之前全部返回.然而,步骤函数对于最终的皱纹是有用的 - 我想在API后面隐藏所有这些.

根据我的阅读,API可以触发lambda并返回该lambda的结果,但我不希望被调用的lambda是返回结果的lambda.当您从API调用步骤函数时,不是API调用返回最后一个状态的结果.

总之,我想:

API请求 - >并行计算结果 - > API响应

中间的那一点我不清楚该怎么做,同时能够将所有结果作为对原始请求的响应返回 - 或者单独使用它们很容易.

我可以看到几个选项:

使用现在由AWS API网关本机支持的步骤函数,并在一个状态中调用多个lambda,等待它们在转换之前返回.

使用AWS EMR,但以某种方式保持配置的实例始终处于活动状态以避免配置时间开销.这显然否定了Lambda的可扩展性并且更加昂贵.

使用mapreduce框架或类似的东西,找到一种方法来响应来自不同lambda的传入请求到API请求最初调用的请求.理想情况下还要减少此处涉及的S3事件的数量,但这不是优先事项.

立即响应来自第一个lambda的原始API请求,然后在计算完成后将更多数据推送给用户(它们应该只需要大约30秒的并行性,并且域是这样的,这是可以接受的等待时间响应,甚至HTTP响应).

我怀疑它会对解决方案产生任何影响,因为它只是中间位的扩展,而不是根本的改变,但真正的计算是迭代的,所以会是:

请求 - > Mapreduce - > Mapreduce - > ... - >响应

只要我知道如何在一个请求中链接一组lambda函数,链接更多应该更多相同(我希望).

谢谢.

PS我无法创建它们,标签aws-emr也不aws-elastic-mapreduce存在.

是否可以有以下类型的阶跃函数图，即来自 2 个并行状态输出，一个组合状态：

如果是，那么 json 会是什么样子？如果不是，为什么？