如何基于前一步执行结果配置AWS StepFunction的执行流程？

该分支流程完全可以实现，核心依赖StepFunction的Choice状态做分支判断，配合Glue任务返回自定义参数即可完成逻辑控制。

1. 改造Glue Python脚本输出判断结果 #

在Glue的Python代码中，完成S3文件写入后，追加空文件判断逻辑，最后将判断结果作为Glue任务的输出返回给StepFunction，示例代码片段：

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

# 原有生成文件、写入S3的业务逻辑
target_bucket = "你的存储桶名称"
target_key = "生成文件的S3路径/xxx.csv"

# 判断文件是否为空
resp = s3.head_object(Bucket=target_bucket, Key=target_key)
file_size = resp['ContentLength']
is_file_empty = file_size == 0

# 向StepFunction返回判断结果（Glue 4.0+支持直接返回JSON格式输出）
print(f'{{"is_file_empty": {str(is_file_empty).lower()}, "s3_file_path": "s3://{target_bucket}/{target_key}"}}')

注意：Glue任务需要开启EnableJobRunInsights配置，才能将标准输出的JSON内容传递给StepFunction

2. 设计StepFunction状态机分支逻辑 #

在状态机中，Glue任务执行完成后添加Choice状态，根据返回的is_file_empty参数做分支：

• 值为true：直接跳转到Succeed状态，标记执行成功
• 值为false：继续执行后续Lambda相关步骤
  示例ASL（Amazon States Language）片段：

{
  "Comment": "带空文件判断的流程示例",
  "StartAt": "运行Glue任务",
  "States": {
    "运行Glue任务": {
      "Type": "Task",
      "Resource": "arn:aws:states:::glue:startJobRun.sync",
      "Parameters": {
        "JobName": "你的Glue任务名"
      },
      "ResultSelector": {
        "is_file_empty.$": "$.JobRun.Output.is_file_empty",
        "s3_file_path.$": "$.JobRun.Output.s3_file_path"
      },
      "Next": "判断文件是否为空"
    },
    "判断文件是否为空": {
      "Type": "Choice",
      "Choices": [
        {
          "Variable": "$.is_file_empty",
          "BooleanEquals": true,
          "Next": "执行成功"
        },
        {
          "Variable": "$.is_file_empty",
          "BooleanEquals": false,
          "Next": "运行Lambda函数"
        }
      ]
    },
    "运行Lambda函数": {
      "Type": "Task",
      "Resource": "arn:aws:states:::lambda:invoke",
      "Parameters": {
        "FunctionName": "你的Lambda函数名",
        "Payload": {
          "s3_file_path.$": "$.s3_file_path"
        }
      },
      "Next": "执行成功"
    },
    "执行成功": {
      "Type": "Succeed"
    }
  }
}

3. Terraform资源配置示例 #

Glue任务配置片段 #

resource "aws_glue_job" "data_gen_job" {
  name     = "data-gen-job"
  role_arn = aws_iam_role.glue_job_role.arn
  command {
    script_location = "s3://你的脚本存储桶路径/glue_script.py"
    python_version  = "3"
  }
  glue_version = "4.0"
  execution_property {
    max_concurrent_runs = 1
  }
  # 开启输出传递配置
  default_arguments = {
    "--enable-job-run-insights" = "true"
  }
}

StepFunction配置片段 #

resource "aws_sfn_state_machine" "data_process_flow" {
  name     = "data-process-flow"
  role_arn = aws_iam_role.sfn_role.arn
  definition = jsonencode({
    Comment = "带空文件判断的流程示例"
    StartAt = "运行Glue任务"
    States = {
      # 填入上文的ASL状态配置即可
    }
  })
}

权限注意事项 #

• Glue任务对应的IAM角色需要添加S3的s3:GetObject、s3:HeadObject权限，授权范围覆盖目标存储桶
• StepFunction对应的IAM角色需要添加Glue的glue:StartJobRun、glue:GetJobRun权限，以及Lambda的lambda:InvokeFunction权限

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