google-cloud-platform - Airflow DAG 未在 Google Cloud Composer 上运行 : "Dependencies Blocking Task From Getting Scheduled"

Question

我刚刚在 Python 3 和 Composer 镜像版本 composer-1.4.0-airflow-1.10.0 上设置了一个 Cloud Composer 环境。所有设置都是“库存”；即没有配置覆盖。

我正在尝试测试一个极其简单的 DAG。它在我的本地 Airflow 服务器上运行没有问题，但在 Cloud Composer 上，Web 服务器的任务信息 View 显示消息 Dependencies Blocking Task From Getting Scheduled

依赖项未知，原因如下:

All dependencies are met but the task instance is not running. In most cases this just means that the task will probably be scheduled soon unless:
- The scheduler is down or under heavy load
- The following configuration values may be limiting the number of queueable processes: parallelism, dag_concurrency, max_active_dag_runs_per_dag, non_pooled_task_slot_count

If this task instance does not start soon please contact your Airflow administrator for assistance.

无论任务是按计划运行，还是当我在 Web 服务器中手动触发它时(我在执行此操作之前将所有任务实例设置为成功，以避免延迟)，都会发生这种情况。我试过resetting the scheduler in kubernetes as per this answer ，但任务仍停留在计划中。

此外，我注意到在我的本地实例(在不同的 Docker 容器上运行服务器、工作程序和调度程序)上，Task Instances View 中的 Hostname 列已填充，但在 Cloud Composer 上，它不是。

这是我正在运行的 DAG:

from datetime import datetime, timedelta
import random

from airflow import DAG
from airflow.operators.python_operator import PythonOperator


default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'email_on_failure': False,
    'email_on_retry': False,
    'queue': 'airflow',
    'start_date': datetime.today() - timedelta(days=2),
    'schedule_interval': None,
    'retries': 2,
    'retry_delay': timedelta(seconds=15),
    'priority_weight': 10,
}


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


def always_succeed():
    pass


always_succeed_operator = PythonOperator(
    dag=example_dag,
    python_callable=always_succeed,
    task_id='always_succeed'
)


def might_fail():
    return 1 / random.randint(0, 1)


might_fail_operator = PythonOperator(
    dag=example_dag, python_callable=might_fail, task_id='might_fail'
)


might_fail_operator.set_upstream(always_succeed_operator)

Answer 1

Cloud Composer 不支持多个 celery 队列，请从默认参数中删除 'queue' : 'airflow'。那应该可以解决您的问题。