AWS MWAA中用Airflow启动GlueCrawler报CrawlerRunningException的解决方法

你看到的Broken DAG并不是任务运行时失败，而是DAG加载阶段就出错了，核心原因是代码执行时机完全不对：

• 你把AwsGlueCrawlerHook的实例化和start_crawler调用直接写在了DAG文件的全局作用域，没有包裹进Airflow的任务逻辑里
• Airflow的Scheduler、Web服务每一次加载解析DAG文件，都会从上到下执行所有全局Python代码，也就是说你的这段启动Crawler的代码，根本不会等DAG到点调度、也不会等上游json2parquet任务跑完，只要DAG文件被加载，就会立刻真实调用AWS API启动Crawler
• 只要加载DAG的当下，目标Crawler已经在运行，API就会抛出CrawlerRunningException，直接导致整个DAG解析失败，弹出Broken DAG提示

不要自己裸调Hook写启动逻辑，也不需要刻意去翻Hook的所有API方法搞清楚每个调用的用途，直接使用AWS provider封装好的专用Operator即可，这个组件已经把启动、状态检查、轮询等待、异常兼容的全流程都封装好了，刚好匹配你“上游任务成功后启动Crawler、运行结束后触发下游”的核心需求。

具体代码写法 #

1. 导入对应的Operator：
   from airflow.providers.amazon.aws.operators.glue_crawler import AwsGlueCrawlerOperator
2. 按业务定义任务实例，必须开启等待完成的参数：

   run_crawler = AwsGlueCrawlerOperator(
       task_id="run_housing_raw_crawler",
       config={"Name": "housing-raw-crawler-crawler-b3be889"},
       aws_conn_id="aws_default", # 替换为你MWAA里配置的可用AWS连接ID
       wait_for_completion=True, # 核心配置：会轮询Crawler状态，直到运行结束才标记任务成功，不会发完启动请求就立刻结束
       poke_interval=30, # 每30秒查询一次Crawler状态，默认值是60秒，可按需调整
   )
   

3. 保持原有任务流转关系即可：
   json2parquet >> run_crawler >> parquet2redshift

针对Crawler运行中报错的兼容逻辑 #

这个Operator的内部实现已经处理了你遇到的CrawlerRunningException场景：

• 发起启动请求前，会先检查Crawler的当前状态，如果已经是运行中，就不会重复发起启动请求，直接进入等待完成的轮询流程
• 等Crawler运行完毕回到READY状态，才会标记任务成功，自动触发下游的parquet2redshift任务

如果有特殊逻辑必须自己基于Hook写，绝对不能把Hook调用放在DAG全局作用域，必须把所有和AWS交互的代码包裹在PythonOperator的执行函数里，只有当任务真正被调度运行时才触发执行，示例代码：

import time
def run_crawler_logic(**context):
    # 仅在任务实际运行时才初始化Hook、调用AWS API
    hook = AwsGlueCrawlerHook(aws_conn_id="aws_default")
    crawler_name = "housing-raw-crawler-crawler-b3be889"
    # 先检查状态，避免重复启动
    if hook.get_crawler_state(name=crawler_name)["State"] == "READY":
        hook.start_crawler(crawler_name=crawler_name)
    # 轮询等待直到运行完成
    while True:
        current_state = hook.get_crawler_state(name=crawler_name)["State"]
        if current_state == "READY":
            break
        time.sleep(30)

from airflow.operators.python import PythonOperator
run_crawler = PythonOperator(
    task_id="run_housing_raw_crawler",
    python_callable=run_crawler_logic
)

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