Step Function 中的 AWS Batch 作业执行结果

如何解决Step Function 中的 AWS Batch 作业执行结果

我是 AWS Step Functions 和 AWS Batch 的新手。我正在尝试将 AWS Batch 作业与 Step Function 集成。 AWS Batch Job 执行输出字符串值的简单 python 脚本（高级简化要求）。我需要让 python 脚本输出可用于 step 函数的下一个状态。我应该如何才能做到这一点。 AWS Batch Job 输出不包含 python 脚本的结果。相反，它包含带有输入值的所有容器相关信息。

示例：AWS Batch 作业执行输出“Hello World”的 python 脚本。我需要“Hello World”可用于 step 函数的下一个状态，以执行与其关联的 lambda。

解决方法

我能够做到，下面是我的状态机，我采用了用于运行批处理作业 Manage a Batch Job (AWS Batch,Amazon SNS) 的示例项目，并将其修改为两个用于传递输入/输出的 lambda。

{
  "Comment": "An example of the Amazon States Language for notification on an AWS Batch job completion","StartAt": "Submit Batch Job","TimeoutSeconds": 3600,"States": {
    "Submit Batch Job": {
      "Type": "Task","Resource": "arn:aws:states:::batch:submitJob.sync","Parameters": {
        "JobName": "BatchJobNotification","JobQueue": "arn:aws:batch:us-east-1:1234567890:job-queue/BatchJobQueue-737ed10e7ca3bfd","JobDefinition": "arn:aws:batch:us-east-1:1234567890:job-definition/BatchJobDefinition-89c42b1f452ac67:1"
      },"Next": "Notify Success","Catch": [
        {
          "ErrorEquals": [
            "States.ALL"
          ],"Next": "Notify Failure"
        }
      ]
    },"Notify Success": {
      "Type": "Task","Resource": "arn:aws:lambda:us-east-1:1234567890:function:readcloudwatchlogs","Parameters": {
        "LogStreamName.$": "$.Container.LogStreamName"
      },"ResultPath": "$.lambdaOutput","Next": "ConsumeLogs"
    },"ConsumeLogs": {
      "Type": "Task","Resource": "arn:aws:lambda:us-east-1:1234567890:function:consumelogs","Parameters": {
        "randomstring.$": "$.lambdaOutput.logs"
      },"End": true
    },"Notify Failure": {
      "Type": "Task","Resource": "arn:aws:states:::sns:publish","Parameters": {
        "Message": "Batch job submitted through Step Functions failed","TopicArn": "arn:aws:sns:us-east-1:1234567890:StepFunctionsSample-BatchJobManagement17968f39-e227-47ab-9a75-08a7dcc10c4c-SNSTopic-1GR29R8TUHQY8"
      },"End": true
    }
  }
}

读取日志的关键在包含 Submit Batch Job 的 LogStreamName 输出中，我将其传递给名为 function:readcloudwatchlogs 的 lambda 并读取日志，然后最终将读取日志传递给下一个名为 function:consumelogs 的函数。您可以在附加的屏幕截图中看到 consumelogs 函数打印日志。

{
  "Attempts": [
    {
      "Container": {
        "ContainerInstanceArn": "arn:aws:ecs:us-east-1:1234567890:container-instance/BatchComputeEnvironment-4a1593ce223b3cf_Batch_7557555f-5606-31a9-86b9-83321eb3e413/6d11fdbfc9eb4f40b0d6b85c396bb243","ExitCode": 0,"LogStreamName": "BatchJobDefinition-89c42b1f452ac67/default/2ad955bf59a8418893f53182f0d87b4b","NetworkInterfaces": [],"TaskArn": "arn:aws:ecs:us-east-1:1234567890:task/BatchComputeEnvironment-4a1593ce223b3cf_Batch_7557555f-5606-31a9-86b9-83321eb3e413/2ad955bf59a8418893f53182f0d87b4b"
      },"StartedAt": 1611329367577,"StatusReason": "Essential container in task exited","StoppedAt": 1611329367748
    }
  ],"Container": {
    "Command": [
      "echo","Hello world"
    ],"ContainerInstanceArn": "arn:aws:ecs:us-east-1:1234567890:container-instance/BatchComputeEnvironment-4a1593ce223b3cf_Batch_7557555f-5606-31a9-86b9-83321eb3e413/6d11fdbfc9eb4f40b0d6b85c396bb243","Environment": [
      {
        "Name": "MANAGED_BY_AWS","Value": "STARTED_BY_STEP_FUNCTIONS"
      }
    ],"Image": "137112412989.dkr.ecr.us-east-1.amazonaws.com/amazonlinux:latest","TaskArn": "arn:aws:ecs:us-east-1:1234567890:task/BatchComputeEnvironment-4a1593ce223b3cf_Batch_7557555f-5606-31a9-86b9-83321eb3e413/2ad955bf59a8418893f53182f0d87b4b",..
  },..
  "Tags": {
    "resourceArn": "arn:aws:batch:us-east-1:1234567890:job/d36ba07a-54f9-4acf-a4b8-3e5413ea5ffc"
  }
}

• 读取日志 Lambda 代码：

import boto3

client = boto3.client('logs')

def lambda_handler(event,context):
    print(event)
    response = client.get_log_events(
        logGroupName='/aws/batch/job',logStreamName=event.get('LogStreamName')
    )
    log = {'logs': response['events'][0]['message']}
    return log

• 使用日志 Lambda 代码

import json

print('Loading function')


def lambda_handler(event,context):
    print(event)