Воздушный поток не планирует корректно Python

Код:

Версия Python 2.7.x и версия 1.5.1 airflow

мой сценарий это

from airflow import DAG
from airflow.operators import BashOperator
from datetime import datetime, timedelta


default_args = {
'owner': 'xyz',
'depends_on_past': False,
'start_date': datetime(2015,10,13),
'email': ['[email protected]'],
'schedule_interval':timedelta(minutes=5),
'email_on_failure': True,
'email_on_retry': True,
'retries': 1,
'retry_delay': timedelta(minutes=5),
}
dag = DAG('testing', default_args=default_args)
run_this_first = BashOperator(task_id='Start1',bash_command='date', dag=dag)
for i in range(5):
    t = BashOperator(task_id="Orders1"+str(i), bash_command='sleep 5',dag=dag)
    t.set_upstream(run_this_first)

Из этого видно, что я создаю группу обеспечения доступности баз данных с 6 задачами, первая задача (Start1) запускается первой, после чего запускаются все остальные пять задач.

В настоящее время я дал 5 минутную задержку между запуском DAG

Он отлично выполнил все шесть заданий первого типа, но через пять минут группа обеспечения доступности баз данных не перезапускается

Прошло более 1 часа, но DAG не был повторно инициирован. Я действительно не знаю, где я был неправ.

Было бы очень хорошо, если бы кто-то мог указать мне, что не так. Я попытался очистить с помощью airflow testing clear затем произойдет то же самое. Он запустил первый экземпляр, а затем просто стоял там.

Единственное, что показывает командная строка, это Getting all instance for DAG testing

Когда я изменяю позицию schedule_interval, он просто запускается без какого-либо параллельного интервала расписания. Это происходит за 5 минут. 300 или более экземпляров задачи завершены. Нет 5-минутного интервала

Код 2:

from airflow import DAG
from airflow.operators import BashOperator
from datetime import datetime, timedelta


default_args = {
'owner': 'xyz',
'depends_on_past': False,
'start_date': datetime(2015,10,13),
'email': ['[email protected]'],
'email_on_failure': True,
'email_on_retry': True,
'retries': 1,
'retry_delay': timedelta(minutes=5),
}
dag = DAG('testing',schedule_interval=timedelta(minutes=5),default_args=default_args)#Schedule here
run_this_first = BashOperator(task_id='Start1',bash_command='date', dag=dag)
for i in range(5):
    t = BashOperator(task_id="Orders1"+str(i), bash_command='sleep 5',dag=dag)
    t.set_upstream(run_this_first)

Ответы

Ответ 1

Для кода 2, я думаю, причина, по которой она запускается каждую минуту:

  • Время начала - 2015-10-13 00:00

  • Интервал расписания составляет 5 минут

  • Каждое сердцебиение планировщика (по умолчанию 5 секунд), ваша DAG будет проверяться

    • Первая проверка: дата начала (нет последней даты исполнения) + планировщик интервал < Текущее время? Если да, то DAG будет исполнена и последняя время записи будет записано. (например, 2015-10-13 00:00 + 5 мин < ток?)
    • Вторая проверка следующего биения: последнее время выполнения + планировщик интервал < Текущее время? Если это так, DAG снова будет выполнена.
    • ....

В решении задано значение start_date DAG как datetime.now() - schedule_interval.

А также, если вы хотите отлаживать:

  • Установка LOGGINGLEVEL в debug в settings.py

  • Измените метод класса is_queueable() airflow.models.TaskInstance на

:

def is_queueable(self, flag_upstream_failed=False):
    logging.debug('Checking whether task instance is queueable or not!')
    if self.execution_date > datetime.now() - self.task.schedule_interval:
        logging.debug('Too early to execute: execution_date {0} + task.schedule_interval {1} > datetime.now() {2}'.format(self.execution_date, self.task.schedule_interval, datetime.now()))
        return False
        ...

Ответ 2

Поскольку время начала (2015-10-13 00:00) меньше, чем сейчас, оно запускает поток backfill. Он будет работать с 2015-10-13 00:00, когда каждую секунду обнаружен планировщик воздушного потока (его дата начала), но Дата выполнения - от 5 минут (время интервала задачи).

См. имя журнала:

$tree airflow/logs/testing/
testing/
|-- Orders10
|   |-- 2015-10-13T00:00:00
|   |-- 2015-10-13T00:05:00
|   -- 2015-10-13T00:10:00
|-- Orders11
|   |-- 2015-10-13T00:00:00
|   |-- 2015-10-13T00:05:00
|   -- 2015-10-13T00:10:00
|-- Orders12
|   |-- 2015-10-13T00:00:00
|   |-- 2015-10-13T00:05:00
|   -- 2015-10-13T00:10:00
|-- Orders13
|   |-- 2015-10-13T00:00:00
|   |-- 2015-10-13T00:05:00
|   -- 2015-10-13T00:10:00
|-- Orders14
|   |-- 2015-10-13T00:00:00
|   |-- 2015-10-13T00:05:00
|   -- 2015-10-13T00:10:00
-- Start1
    |-- 2015-10-13T00:00:00
    |-- 2015-10-13T00:05:00
    |-- 2015-10-13T00:10:00
    -- 2015-10-13T00:15:00

См. время создания журналов:

$ll airflow/logs/testing/Start1
-rw-rw-r-- 1 admin admin 4192 Nov  9 14:50 2015-10-13T00:00:00
-rw-rw-r-- 1 admin admin 4192 Nov  9 14:50 2015-10-13T00:05:00
-rw-rw-r-- 1 admin admin 4192 Nov  9 14:51 2015-10-13T00:10:00
-rw-rw-r-- 1 admin admin 4192 Nov  9 14:52 2015-10-13T00:15:00

Кроме того, вы можете увидеть экземпляры задач в веб-интерфейсе:

Экземпляры задач потока воздуха