Интерфейс воздушного потока не может вызвать dag вручную - PullRequest
Новая
       8

Интерфейс воздушного потока не может вызвать dag вручную

0 голосов
/ 03 декабря 2018

Версии:

  • Airflow v1.10.1
  • Сервер SQL Server
  • Веб-сервер и планировщик, работающие на одном хосте
  • LocalExecutor

Простой DAG world hello world не запускается, когда запускается вручную с помощью пользовательского интерфейса (кнопка).Тот же пример работает нормально при запуске через командную строку.Хотелось бы разрешить пользователям использовать пользовательский интерфейс для запуска задания.Это ошибка.

Пример Hello world DAG проверено: 

from datetime import datetime
from airflow import DAG
from airflow.operators.dummy_operator import DummyOperator
from airflow.operators.python_operator import PythonOperator

def print_welcome():
    return 'Welcome!'

dag = DAG('say_welcome', description='Simple tutorial DAG',
          schedule_interval='0 12 * * *',
          start_date=datetime(2017, 3, 20), catchup=False)

dummy_operator = DummyOperator(task_id='say_welcome_dummy_task', retries=3, dag=dag)

hello_operator = PythonOperator(task_id='say_welcome_task', python_callable=print_welcome, dag=dag)

dummy_operator >> hello_operator

Проверка вывода из командной строки.

(airfow_v1_venv) sshuser@ed41-kp06sp:~/airflowv1/dags$ airflow trigger_dag say_welcome
[2018-12-03 19:38:34,679] {__init__.py:51} INFO - Using executor LocalExecutor
[2018-12-03 19:38:34,956] {models.py:271} INFO - Filling up the DagBag from /home/sshuser/airflowv1/dags
[2018-12-03 19:38:35,071] {cli.py:241} INFO - Created <DagRun say_welcome @ 2018-12-03 19:38:34+00:00: manual__2018-12-03T19:38:34+00:00, externally triggered: True>

Журналы при запуске с использованием пользовательского интерфейса 

    context)
  File "/home/sshuser/airfow_v1_venv/local/lib/python2.7/site-packages/sqlalchemy/engine/default.py", line 467, in do_executemany
    cursor.executemany(statement, parameters)
IntegrityError: (pyodbc.IntegrityError) ('23000', u"[23000] [Microsoft][ODBC Driver 17 for SQL Server][SQL Server]Violation of PRIMARY KEY constraint 'PK__task_ins__9BEABD04E2A8D429'. Cannot insert duplicate key in object 'dbo.task_instance'. The duplicate key value is (say_welcome_task, say_welcome, Dec  3 2018  7:40PM). (2627) (SQLExecDirectW)") [SQL: u'INSERT INTO task_instance (task_id, dag_id, execution_date, start_date, end_date, duration, state, try_number, max_tries, hostname, unixname, job_id, pool, queue, priority_weight, operator, queued_dttm, pid, executor_config) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)'] [parameters: (('say_welcome_task', 'say_welcome', datetime.datetime(2018, 12, 3, 19, 40, 9, 787000, tzinfo=<Timezone [UTC]>), None, None, None, None, 0, 0, u'', 'sshuser', None, None, 'default', 1, None, None, None, bytearray(b'\x80\x02}q\x00.')), ('say_welcome_dummy_task', 'say_welcome', datetime.datetime(2018, 12, 3, 19, 40, 9, 787000, tzinfo=<Timezone [UTC]>), None, None, None, None, 0, 3, u'', 'sshuser', None, None, 'default', 2, None, None, None, bytearray(b'\x80\x02}q\x00.')))]

1 Ответ

0 голосов
/ 04 декабря 2018

Похоже, вы пытались переопределить старый say_welcome DAG.

Создайте один с именем say_welcome_v1 и сделайте ему снимок.

Когда создается новый DAG, вынужно изменить его имя, чтобы его можно было различить в метабазе.Следовательно, соглашение об использовании _v1, _v2 и т. Д. В конце имени группы обеспечения доступности баз данных каждый раз, когда происходят изменения в группе обеспечения доступности баз данных.

Поскольку полученная вами ошибка является ошибкой целостности, которая возникает, когдаВы пытаетесь вставить что-то в БД с тем же первичным ключом, что и другой, уже существующий.Скорее всего, ошибка нового DAG с тем же именем, что и у старого.

Если у вас нет старых прогонов DAG, историю / журналы которых стоит хранить, вы можете просто использовать airflow resetdb с последующимairflow initdb для сброса базы данных и запуска с нуля.

Вы также можете удалить старый DAG ID из метабазы ​​данных, используя airflow delete_dag my_dag_id начиная с версии 1.10 Airflow и выше.

...