使用AWS Step Functions调用ECSProperties类型AWS Batch Job失败求助

问题概述 #

工作流逻辑为：Lambda准备AWS Batch作业定义（Fargate类型，基于ecsProperties），随后通过Step Functions的arn:aws:states:::batch:submitJob.sync步骤提交作业并等待完成。但执行时触发400错误：

Container overrides should not be set for ecsProperties jobs. (Service: AWSBatch; Status Code: 400; Error Code: ClientException)

关键差异：

• 直接在Lambda中用boto3的submit_job API提交作业可正常运行
• Step Functions的同步提交步骤会自动注入以下ContainerOverrides配置，这是报错根源：

"ContainerOverrides":{
  "Environment": [
    {
      "Name": "MANAGED_BY_AWS",
      "Value": "STARTED_BY_STEP_FUNCTIONS"
    }
  ]
}

必须使用Step Functions等待作业完成的原因：Lambda存在15分钟超时限制，且boto3的submit_job仅提交作业不等待执行完成，无法满足“Batch作业完成后插入Athena分区”的业务需求。

解决方案：用Step Functions轮询Batch作业状态 #

由于Step Functions的submitJob.sync集成会强制添加不被Fargate类型Batch作业支持的ContainerOverrides，我们可以通过Lambda提交作业+Step Functions轮询状态的方式替代，既规避报错，又实现等待作业完成的逻辑。

步骤1：修改Lambda提交作业并返回JobId #

调整原Lambda函数，提交作业后返回jobId给Step Functions，不要设置containerOverrides参数：

import boto3

batch_client = boto3.client('batch')

def lambda_handler(event, context):
    # 保留原作业准备逻辑
    job_def = "your-fargate-job-definition"
    job_queue = "your-fargate-job-queue"
    job_name = f"batch-job-{context.aws_request_id}"

    # 提交Fargate类型Batch作业
    submit_response = batch_client.submit_job(
        jobName=job_name,
        jobQueue=job_queue,
        jobDefinition=job_def
        # 按需添加其他作业参数，禁止添加containerOverrides
    )

    return {"jobId": submit_response["jobId"]}

步骤2：Step Functions工作流添加轮询逻辑 #

将原Run Batch Job步骤替换为循环轮询结构，通过AWS SDK集成查询作业状态，直到作业完成或失败：

工作流定义示例片段：

{
  "States": {
    "Submit Batch Job": {
      "Type": "Task",
      "Resource": "arn:aws:states:::lambda:invoke",
      "Parameters": {
        "FunctionName": "your-submit-job-lambda",
        "Payload.$": "$"
      },
      "ResultPath": "$.batchJob",
      "Next": "Check Job Status"
    },
    "Check Job Status": {
      "Type": "Task",
      "Resource": "arn:aws:states:::aws-sdk:batch:describeJobs",
      "Parameters": {
        "jobs.$": ["$.batchJob.jobId"]
      },
      "ResultPath": "$.jobStatus",
      "Next": "Evaluate Job Status"
    },
    "Evaluate Job Status": {
      "Type": "Choice",
      "Choices": [
        {
          "Variable": "$.jobStatus.jobs[0].status",
          "StringEquals": "SUCCEEDED",
          "Next": "Insert Athena Partition"
        },
        {
          "Variable": "$.jobStatus.jobs[0].status",
          "StringEquals": "FAILED",
          "Next": "Mark Job Failed"
        }
      ],
      "Default": "Wait and Retry"
    },
    "Wait and Retry": {
      "Type": "Wait",
      "Seconds": 60, // 根据作业时长调整等待间隔
      "Next": "Check Job Status"
    },
    "Mark Job Failed": {
      "Type": "Fail",
      "Cause": "Batch job execution failed",
      "Error": "BatchJobFailure"
    },
    "Insert Athena Partition": {
      "Type": "Task",
      // 配置执行Athena分区插入的逻辑（如Lambda或Athena集成）
      "End": true
    }
  }
}

方案优势 #

• 彻底规避Step Functions自动添加ContainerOverrides的问题，Batch作业可正常提交
• Step Functions标准工作流无15分钟超时限制（最长可执行1年），支持等待长时Batch作业
• 完整保留“Batch完成后执行Athena操作”的业务逻辑

内容的提问来源于stack exchange，提问作者Emile Beukes