如何让AWS Step Functions等待Glue Crawler执行完成？

这个问题我之前帮团队落地过，太懂轮询Lambda那种低效又费钱的痛点了！用CloudWatch Events配合Step Functions的回调模式才是最优解，下面一步步给你讲怎么实现：

核心思路 #

你的Step Functions流程会先启动Glue Crawler，然后进入等待回调状态；当Crawler执行完成（成功/失败）时，CloudWatch Events会自动捕获到这个状态变更事件，触发回调动作让Step Functions继续往下走——完全不用轮询，资源利用率拉满。

步骤1：配置CloudWatch Events规则，捕获Crawler状态变更 #

先去CloudWatch控制台创建一个事件规则：

• 事件源选AWS Glue，事件类型挑Crawler State Change
• 指定你要监控的Crawler名称（支持通配符，方便批量管理多个Crawler）
• 只筛选SUCCEEDED和FAILED这两个状态（毕竟只有这两种情况代表Crawler执行完成）
• 目标设置为一个中转Lambda函数（后面会写这个函数的逻辑，它负责把事件信息转成Step Functions能识别的回调信号）

步骤2：写中转Lambda函数，处理回调逻辑 #

这个Lambda的作用很简单：拿到CloudWatch Event里的Crawler状态，找到对应的Step Functions执行的TaskToken，然后调用Step Functions的回调API让流程继续。

这里需要用DynamoDB存一下TaskToken和Crawler名称的关联（Step Functions启动Crawler时会把这个关系存进去）。给你个Python示例：

import boto3
import json

dynamodb = boto3.client('dynamodb')
stepfunctions = boto3.client('stepfunctions')

def lambda_handler(event, context):
    # 从CloudWatch Event里抠出关键信息
    crawler_name = event['detail']['crawlerName']
    crawler_state = event['detail']['state']
    
    # 从DynamoDB里取出之前存的TaskToken
    token_response = dynamodb.get_item(
        TableName='CrawlerTaskTokens',
        Key={'CrawlerName': {'S': crawler_name}}
    )
    task_token = token_response['Item']['TaskToken']['S']
    
    # 根据Crawler状态调用不同的回调API
    if crawler_state == 'SUCCEEDED':
        stepfunctions.send_task_success(
            taskToken=task_token,
            output=json.dumps({'CrawlerName': crawler_name, 'Status': 'Done'})
        )
    else:
        stepfunctions.send_task_failure(
            taskToken=task_token,
            error='CrawlerFailed',
            cause=f"Crawler {crawler_name} 执行失败，状态：{crawler_state}"
        )
    
    # 用完就删，避免重复触发回调
    dynamodb.delete_item(
        TableName='CrawlerTaskTokens',
        Key={'CrawlerName': {'S': crawler_name}}
    )
    
    return {'statusCode': 200, 'message': '回调处理完成'}

步骤3：设计Step Functions状态机流程 #

状态机的关键是要生成TaskToken并存储，然后进入等待状态。给你个核心片段参考：

{
  "States": {
    "启动数据源Crawler": {
      "Type": "Task",
      "Resource": "arn:aws:states:::aws-sdk:glue:startCrawler",
      "Parameters": {
        "Name.$": "$.SourceCrawlerName"
      },
      "Next": "存储TaskToken"
    },
    "存储TaskToken": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:你的区域:你的账号:function:存Token的Lambda",
      "Parameters": {
        "CrawlerName.$": "$.SourceCrawlerName",
        "TaskToken.$": "$$.Task.Token"
      },
      "Next": "等待Crawler完成"
    },
    "等待Crawler完成": {
      "Type": "Task",
      "Resource": "arn:aws:states:::lambda:invoke.waitForTaskToken",
      "Parameters": {
        "FunctionName": "随便整个占位Lambda就行", // 实际不会执行，靠回调触发
        "Payload": {
          "TaskToken.$": "$$.Task.Token"
        }
      },
      "TimeoutSeconds": 3600, // 根据你的Crawler实际运行时间设超时
      "Next": "执行Glue Job",
      "Catch": [
        {
          "ErrorEquals": ["States.Timeout"],
          "Next": "Crawler超时处理"
        },
        {
          "ErrorEquals": ["CrawlerFailed"],
          "Next": "Crawler失败处理"
        }
      ]
    },
    // 后面就是执行Glue Job、启动目标Crawler的步骤，和你原来的逻辑一致
  }
}

步骤4：配置IAM权限（别漏了！） #

给相关资源加对权限：

• Step Functions要有调用Glue.StartCrawler和Lambda的权限
• 中转Lambda要有DynamoDB读写和StepFunctions.SendTaskSuccess/SendTaskFailure的权限
• CloudWatch Events要有调用中转Lambda的权限

简化版小技巧 #

如果你的Crawler和Step Functions执行是一一对应的，也可以不用DynamoDB，直接在Step Functions启动Crawler时把执行ARN和Crawler名称存在环境变量里，但DynamoDB的方式更通用，适合多并发的场景。

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