Airflow任务卡在queued状态：依赖阻碍任务调度问题排查

1. 修复DAG重复定义的问题 #

你的代码里重复定义了两个同名DAG（dag_id='upload_to_s3'），第一个DAG块是空的（仅含pass），第二个才包含上传任务。Airflow加载时会出现识别冲突，实际生效的可能是那个空DAG，导致触发后无任务可执行。

解决方法：删除第一个空的DAG定义块，保留包含PythonOperator的那个DAG，修改后的代码如下：

# airflow related
from airflow import DAG
from airflow.operators.python import PythonOperator
# other packages
from datetime import datetime
import boto3

def file_upload():
    # 创建Boto3会话
    session = boto3.Session(
        aws_access_key_id='my_access_key_id',
        aws_secret_access_key='my_aws_secret_access_key'
    )

    # 创建S3资源实例
    s3 = session.resource('s3')
    result = s3.Bucket('flight-data-test-bucket').upload_file('/opt/airflow/dags/pricedata.xlsx', 'pricedata.xlsx')

    return result

with DAG(
    dag_id='upload_to_s3',
    start_date=datetime(2020, 5, 5),
    schedule_interval='@once',
    catchup=False,
) as dag:
    # 定义上传任务
    task_file_to_s3 = PythonOperator(
        task_id='upload_to_s3',
        python_callable=file_upload
    )

2. 检查Airflow Worker服务状态 #

Docker部署的Airflow中，任务需要Worker节点执行，若Worker未运行或离线，任务会一直卡在queued状态：

• 执行docker ps命令，查看是否有airflow-worker容器处于Up状态；
• 或在Airflow UI的Admin -> Workers页面，确认Worker状态为在线。

若Worker未运行，执行以下命令启动：

docker-compose up -d worker

若集群状态异常，可重启所有服务：

docker-compose down && docker-compose up -d

3. 验证调度配置逻辑 #

你的start_date=datetime(2020,5,5)、schedule_interval='@once'、catchup=False配置本身合理，但手动触发时需注意：

• 手动触发会创建独立的DAG Run，不受start_date影响，但前提是DAG定义正确（已解决第一步的重复定义问题）；
• 若仍有调度异常，可尝试将start_date改为较近的过去日期（如datetime(2024, 1, 1)），避免Airflow调度器的潜在时间逻辑问题。

4. 提前排查执行阶段的潜在问题 #

虽然当前任务卡在Queued阶段，可提前排查后续执行可能遇到的问题：

• 文件路径与权限：确保/opt/airflow/dags/pricedata.xlsx文件存在于Airflow Worker容器中，且Worker用户有读取权限。可进入Worker容器验证：

  docker exec -it <airflow-worker-container-id> bash
  ls /opt/airflow/dags/pricedata.xlsx
  

• AWS密钥安全性：不要在代码中硬编码AWS密钥，建议通过Airflow UI的Admin -> Connections创建S3连接，或通过环境变量传递密钥，避免泄露风险。

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