Airflow worker卡住:任务处于“running”状态，这不是有效的执行状态。必须清除任务才能运行

基础概念

Apache Airflow 是一个用于创建、调度和监控工作流的开源平台。它使用有向无环图（DAG）来定义任务之间的依赖关系。Airflow 的工作流程包括调度器（Scheduler）、Web 服务器（Webserver）和 Worker。

调度器：负责将 DAG 文件加载到内存中，并根据依赖关系触发任务。
Web 服务器：提供用户界面，用于查看和管理 DAG 和任务。
Worker：实际执行任务的进程。

问题描述

当 Airflow worker 卡住，任务处于“running”状态，但无法完成执行时，这通常是由于以下几种原因之一：

任务执行时间过长：任务可能需要很长时间才能完成，导致超时。
资源不足：Worker 所在的机器资源（如 CPU、内存）不足。
任务依赖问题：任务的依赖关系没有正确配置，导致任务无法继续执行。
代码错误：任务代码中存在错误，导致任务无法正常执行。
外部依赖问题：任务依赖于外部服务或数据源，而这些服务或数据源不可用。

解决方法

1. 检查任务日志

首先，查看任务的日志文件，以确定任务卡住的具体原因。日志文件通常位于 Airflow 的日志目录中。

# 查看任务日志
airflow tasks log <task_id>

2. 增加超时时间

如果任务执行时间过长，可以增加任务的超时时间。

from airflow import DAG
from airflow.operators.dummy_operator import DummyOperator
from datetime import timedelta

default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': datetime(2023, 1, 1),
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
    'execution_timeout': timedelta(hours=24)  # 增加超时时间
}

dag = DAG(
    'example_dag',
    default_args=default_args,
    schedule_interval=timedelta(days=1),
)

task = DummyOperator(task_id='dummy_task', dag=dag)

3. 增加资源

如果资源不足，可以考虑增加 Worker 的资源，例如增加 CPU 和内存。

# 增加 Worker 资源
airflow worker --cpu 4 --memory 8G

4. 检查任务依赖关系

确保任务的依赖关系正确配置。

from airflow import DAG
from airflow.operators.dummy_operator import DummyOperator
from datetime import datetime, timedelta

default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': datetime(2023, 1, 1),
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
}

dag = DAG(
    'example_dag',
    default_args=default_args,
    schedule_interval=timedelta(days=1),
)

start_task = DummyOperator(task_id='start', dag=dag)
end_task = DummyOperator(task_id='end', dag=dag)

# 正确配置任务依赖关系
start_task >> end_task

5. 检查代码错误

检查任务代码中是否存在错误，并进行修复。

from airflow import DAG
from airflow.operators.python_operator import PythonOperator
from datetime import datetime, timedelta

def my_function():
    # 确保代码没有错误
    pass

default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': datetime(2023, 1, 1),
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
}

dag = DAG(
    'example_dag',
    default_args=default_args,
    schedule_interval=timedelta(days=1),
)

task = PythonOperator(
    task_id='python_task',
    python_callable=my_function,
    dag=dag,
)

6. 检查外部依赖

确保任务依赖的外部服务或数据源可用。

from airflow import DAG
from airflow.operators.http_operator import SimpleHttpOperator
from datetime import datetime, timedelta

default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': datetime(2023, 1, 1),
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
}

dag = DAG(
    'example_dag',
    default_args=default_args,
    schedule_interval=timedelta(days=1),
)

task = SimpleHttpOperator(
    task_id='http_task',
    method='GET',
    http_conn_id='http_default',
    endpoint='some_endpoint',
    dag=dag,
)