Почему задача застревает и не выполняется в потоке воздуха?

Question

Я пробую воздушный поток с BigQueryOperator. Я думал, что позже буду использовать Google Composer, но сначала я хочу, чтобы он работал локально. У меня есть поток воздуха, и BashOperator работает нормально, я также могу запустить airflow test <dag> <task>, где task - это большая задача запроса, которую я хочу выполнить, но когда я запускаю 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 ..., но не когда он запускается планировщиком или пользовательским интерфейсом? Что-то не так с тем, что у меня здесь есть?

Локально я использую стандартную установку воздушного потока с sqllite, но это не должно иметь никакого влияния, я думаю. Я запускаю все в одном окружении Python, поэтому он должен быть довольно сдержанным.

Answer 1

Если это ваша первая настройка Airflow, вы можете сначала проверить эти вещи: Airflow 1.9.0 ставится в очередь, но не запускает задачи

Кроме того, здесь я бы порекомендовал особенно последний шаг:

• Если больше ничего не работает, вы можете использовать веб-интерфейс, чтобы щелкнуть по метке, а затем в графическом представлении. Теперь выберите первую задачу и нажмите «Экземпляр задачи». В разделе Подробности экземпляра задачи вы увидите, почему группа обеспечения доступности баз данных ожидает или не работает.

Это может дать вам более полное представление о том, почему задача не запланирована.