标签: aws-step-functions

我有一个工作 StepFunction 状态机，有 4 个步骤触发 4 个 Lambda。

由于其中一个步骤将是一个运行时间较长的任务，因此我决定将其中一个 Lambda 转换为 Fargate 任务。

配置 ECS 和 Fargate 任务后，我尝试更新状态机定义，但收到错误：Failed to update state machine.没有任何其他消息。

我的状态机定义似乎是有效的，就是这样，只是没有实际的 ARN：

{
  "Comment": "My Workflow",
  "StartAt": "Step1",
  "States": {
    "Step1": {
      "Type": "Task",
      "Resource": "copy-pasted-arn-of-lambda",
      "Next": "Step2"
    },
    "Step2": {
      "Type": "Task",
      "Resource": "arn:aws:states:::ecs:runTask.sync",
      "Parameters": {
        "LaunchType": "FARGATE",
        "Cluster": "copy-pasted-arn-of-cluster",
        "TaskDefinition": "copy-paster-arn-of-task-definition",
        "Overrides": {
          "ContainerOverrides": [
            {
              "Name": "container-name",
              "Command.$": "$.commands"
            }
          ]
        }
      },
      "Next": "Step3",
      "Catch": [
        {
          "ErrorEquals": [
            "States.ALL"
          ],
          "Next": …

下面的步骤函数在 aws 中执行，当缺少必需参数时，它会取消流程并抛出 States.Runtime 错误。这是处于步骤函数的捕获阶段，但它没有捕获所述错误。

定义的Step函数如下，

{
  "StartAt": "Log Start Step Function",
  "Comment": "Executed with inputs",
  "States": {
    "Log Start Step Function": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:eu-west-1:0000000:function:update",
      "Parameters": {
        "body": {
          "itemID.$": "$.itemID",
          "functionName.$": "$.stepFunctionName ",
          "executionARN.$": "$$.Execution.Id",
          "complete": false,
          "inprogress": true,
          "error": false
        }
      },
      "Catch": [
        {
          "ErrorEquals": [
            "States.Runtime"
          ],
          "ResultPath": "$.taskresult",
          "Next": "Log Failed Module"
        },
        {
          "ErrorEquals": [
            "States.ALL"
          ],
          "ResultPath": "$.taskresult",
          "Next": "Log Failed Module"
        }

      ],
      "ResultPath": "$.taskresult",
      "Next": "Evaluate Module PA1"
    } …

如果我有一个定义 AWS Step Functions 的 CDK *.ts 文件，是否可以生成一个 Amazon States Language asl.json 文件，我可以用它来可视化该步骤函数流程（使用 AWS Toolkit for VS）？

我查看了：CDK 定义时是否有一种在本地运行 AWS Step Functions 的方法？，有没有办法使用CDK创建阶跃函数图？，以及用于 Step Functions 的 AWS CDK：https://docs.aws.amazon.com/cdk/api/latest/docs/aws-stepfunctions-readme.html但这些资源都没有表明生成 asl.json 的过程文件。AWS Step Function 模块有一个导入，我要寻找的本质上是相反的或导出。

我正在执行步进功能。但是，我在执行一种状态时出错。

它给出错误：Lambda.Unknown 原因：无法确定原因，因为 Lambda 没有返回错误类型。

我的 lambda 函数调用外部网络服务。我不想等到收到网络服务响应。

我们可以在不等待 web 服务响应的情况下从函数返回数据吗？？

我的 Lambda 函数

var http = require('http');
exports.handler = (event, context, callback) => {

    var inputJson= {};

    inputJson.firstname= event.firstname; 
    inputJson.lastname= event.lastname;
    inputJson.workspacename= event.workspacename;
    inputJson.customermailid= event.customermailid;
    inputJson.mobilenumber= event.mobilenumber;
    inputJson.orgname= event.orgname;

    inputJson.sessionid= event.sessionid;

    var post_data = JSON.stringify({
        "domainname" : inputJson.domainname,
        "orgname" : inputJson.orgname,
        "customermailid" : inputJson.customermailid,
        "adminmailid":"postmaster@test.int",
        "product":3
    });

  // An object of options to indicate where to post to
  var post_options = {
      host: 'host_ip',
      path: 'path',
      method: 'POST', …

我是 Step Functions 和 AWS 的新手。是否可以从 AWS (EMR) 中的另一个 Step Function 调用一个 Step Function？我正在开发一个 Step 函数，在停止 EMR 之前我必须将 Step 函数包含在 EMR 中。

这里我有两种情况。

1) 我必须从 SFN2 呼叫 SFN3 并停止 SFN2 中的 EMR。2) 执行所有步骤功能后我必须停止 EMR。

我已附上附件供您参考。

请帮助我。

非常感谢。

我使用AWS Step Functions并拥有以下工作流程

initStep - 这是一个 lambda 函数处理程序，它获取一些数据并将其发送给SQS外部服务。

activity = os.getenv('ACTIVITY')
queue_name = os.getenv('QUEUE_NAME')

def lambda_handler(event, context):
  event['my_activity'] = activity
  data = json.dumps(event)

  # Retrieving a queue by its name
  sqs = boto3.resource('sqs')
  queue = sqs.get_queue_by_name(QueueName=queue_name)

  queue.send_message(MessageBody=data, MessageGroupId='messageGroup1' + str(datetime.time(datetime.now())))

  return event

validationWaiting - 它activity等待来自包含数据的外部服务的答案。

complete - 它是一个 lambda 函数处理程序，它使用initStep.

def lambda_handler(event, context):
  email = event['email'] if 'email' in event else None
  data = event['data'] if 'data' in event else None

  client …

我的任务是将依赖节点列表转换为 AWS Step Functions。AWS Step Function 定义允许并行分支甚至分支嵌套到多级深度。不幸的是，它不支持分支中任务之间的依赖关系，因此强制您在两个结果可用于步骤函数中的后续任务之前完成并行步骤。

在我的图表中，Step Functions 可以轻松支持如图 1 所示的简单并行分支。

当涉及到图 2，尤其是图 3 时，它就成了一个问题。

作为一种简单的方法，我们可以引入额外的节点来为其依赖节点收集结果，如图 2b 和 3b 所示，但这现在引入了以前不存在的依赖关系：

• 在图 2b 中，引入了这些新的依赖项：
  • E -> A，F -> A
• 在图 3b 中，引入了这些新的依赖项：
  • E -> A、F -> A、F-> B、C -> E

这是一个问题，因为在手动审批任务的情况下，这些任务的时间可能是几小时到几天。这将导致后面的步骤被它们不依赖的任务不必要地延迟。

有关如何解决此问题的任何建议？也许我可以采取不同的方法？也许我可以应用一些花哨的图论算法？我什至不知道用什么词来解释图论中的这个问题。

如果需要，这里有一个在 draw.io 上使用这些图表的网址。

我创建了一个带有嵌套 Step Functions (SF) 的 POC，并遇到了以下问题。

• 我有一个 SF A，它有一个同步任务，它开始执行 SF B（使用最近宣布的语法）并等待它完成。
• SF B 返回有效的 JSON 输出
• 在SF A 中，我可以看到（从调用SF B 的任务的输出）返回的正确字段。例如状态、ExecutionArn 以及输出。
• 然后我想在 SF A 中执行一个使用从 SF B 返回的字段的选择任务。 但是，从 SF B 返回的输出字段中的 json 是转义的 json，因此 jsonPath 无法从输出数据中提取字段来自嵌套的json。

对于这种情况，我缺少什么或推荐的方法吗？我可以在 lambda 或其他东西中“处理”转义的 json 字符串，但在选择任务中使用它......但这有点违背了嵌套 SF 工作流程的要点。

任何帮助表示赞赏。

目前，我正在研究将复杂的 Web 应用程序从 AWS 迁移到 GCP 的可能性和正确方法。将通用计算和网络服务从一个提供商映射到另一个提供商实际上没有问题，但我想知道 GCP 是否有类似于AWS Step Functions 的服务？我已经看过 Google Dataflow和 Google Cloud Tasks。第二个似乎是这样的，但我不确定它是否是最佳解决方案。

所以问题是谷歌的哪些服务提供了与 AWS Step Functions 相同的功能？如果没有这样的 - 那么你会推荐哪些服务的组合来实现分布式任务（主要是云功能）的有效编排。谢谢！

我正在尝试使用 AWS Step Functions 通过 Lambda 触发对非常大的 S3 文件的操作。为此，我使用具有文件的 S3 键和该文件的字节范围的输入调用步进函数（每个并行迭代将在文件的不同部分进行操作）。输入看起来像

{
  "job-spec": {
    "file": "some_s3_key",
    "array": [
      "0-100",
      "101-200",
      "201-300", ...
    ]
  }
}

我的 Step 函数非常简单，获取输入并将其映射出来，但是我似乎无法将文件和数组都作为我的 lambda 的输入。这是我的步骤函数定义

{
  "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": "$.job-spec",
      "ResultPath": "$.array",
      "MaxConcurrency": 2,
      "Next": "Final State",
      "Iterator": {
        "StartAt": "My Stage",
        "States": {
          "My …