为什么任务被卡住而不在气流中执行？

Question

我正在尝试使用BigQueryOperator。我以为以后会使用google composer，但我希望它先在本地运行。我可以顺畅运行并且运行BashOperator良好，也可以airflow test <dag> <task>在task要运行的大查询任务所在的位置运行，但是当我从UI触发DAG时，bigquery任务永远不会排队。相反，它们具有REMOVED状态，什么也没有发生。

我的DAG定义如下：

from airflow import DAG
from airflow.operators.bash_operator import BashOperator
from datetime import datetime, timedelta
from airflow.contrib.operators.bigquery_operator import BigQueryOperator

yesterday = datetime.combine(
    datetime.today() - timedelta(1),
datetime.min.time())

default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'email': ['airflow@example.com'],
    'start_date': yesterday,
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
}

with DAG(
        'tutorial', default_args=default_args) as dag:

operators
    t1 = BashOperator(
        task_id='print_date',
        bash_command='date')

    template_sql = '''
            SELECT 'TOMAS' name, '{{ params.my_value }}' value, "{{ params.my_value2 }}" tables
        '''

    sampleBigQuery1 = BigQueryOperator(
        task_id='bq_simple_sql1',
        bql=template_sql,
        use_legacy_sql=False,
        destination_dataset_table='temp_tomas.airflow_1',
        allow_large_results=True,
        params={'my_value': (datetime.now()).strftime("%D %T"),
                'my_value2': "yolo"},  # getTables()},
        create_disposition='CREATE_IF_NEEDED',
        write_disposition='WRITE_TRUNCATE'
    )

    t1 >> sampleBigQuery1

那么，如何调试一个能正常运行airflow test ...但无法由调度程序或UI触发的情况？我在这里的东西似乎有问题吗？

我在本地使用sqllite安装标准的airflow，但是我认为这不会产生任何影响。我正在一个python env中运行所有内容，因此应该包含得很好。

Answer 1

如果这是您的第一次 Airflow 设置，您可能需要先检查这些内容：Airflow 1.9.0 正在排队但未启动任务

此外，在这里我特别推荐最后一步：

• 如果没有其他效果，您可以使用 Web ui 单击 dag，然后单击 Graph View。现在选择第一个任务并单击任务实例。在任务实例详细信息段落中，您将看到 DAG 等待或未运行的原因。

这可能会让您更多地了解为什么没有安排任务。