Airflow Dag не перестает работать при сбое скрипта Python - PullRequest
Новая
       10

Airflow Dag не перестает работать при сбое скрипта Python

0 голосов
/ 10 января 2019

Я использую некоторые скрипты Python для загрузки файлов с ftp-сервера, и я создал DAG для этого. Иногда сценарий python при загрузке файла с FTP-сервера завершается с ошибкой «Ошибка сброса соединения с равноправным узлом», но DAG с воздушным потоком не завершается с ошибкой и отмечает задачу как успешную, а не как неудачную. 

Below are airflow logs for more information.


[2019-01-03 19:04:40,085] {base_task_runner.py:98} INFO - Subtask: [2019-01-03 19:04:40,085] {ssh_execute_operator.py:146} INFO - [2019-01-03 19:09:14,276 - Download files from SFTP - ERROR] Total 1 file(s) ([u'R0000797-Manifest.xml']) are downloaded successfully. One error is found in downloading file xxxxxx.txt due to Server connection dropped:
[2019-01-03 19:04:40,091] {base_task_runner.py:98} INFO - Subtask: [2019-01-03 19:04:40,090] {ssh_execute_operator.py:146} INFO - [2019-01-03 19:09:14,282 - Download files from SFTP - ERROR] The whole process failed due to Server connection dropped: .
[2019-01-03 19:04:40,091] {base_task_runner.py:98} INFO - Subtask: [2019-01-03 19:04:40,091] {ssh_execute_operator.py:146} INFO - Total 1 file(s) ([u'R0000797-Manifest.xml']) are downloaded successfully.
[2019-01-03 19:04:40,092] {base_task_runner.py:98} INFO - Subtask: [2019-01-03 19:04:40,091] {ssh_execute_operator.py:146} INFO - Traceback (most recent call last):
[2019-01-03 19:04:40,092] {base_task_runner.py:98} INFO - Subtask: [2019-01-03 19:04:40,091] {ssh_execute_operator.py:146} INFO - main(args)
[2019-01-03 19:04:40,092] {base_task_runner.py:98} INFO - Subtask: [2019-01-03 19:04:40,091] {ssh_execute_operator.py:146} INFO - File "/TEST/GET_files.py", line 381, in main
[2019-01-03 19:04:40,093] {base_task_runner.py:98} INFO - Subtask: [2019-01-03 19:04:40,092] {ssh_execute_operator.py:146} INFO - sftp.get(source_file)


As you can see from above logs that python script gave a proper error message to airflow handler but airflow handler shows that message as INFO and it doesn't fail. So please can you suggest me or help me in this scenario? I want to fail the DAG task when any python error occurs. 

************************************
here is the dag code

get_files = SSHExecuteOperator(
    task_id='get_files',
    bash_command=command to run the py script,
    ssh_hook=sshHook,
    dag=dag)

************************************

Expected results: The airflow DAG should fail when python script fails.
Thanks for your help in advance.

Ответы [ 2 ]

0 голосов
/ 11 января 2019

Почему вы не используете PythonOperator?

Пример здесь: https://github.com/trbs/airflow-examples/blob/master/dags/example_python_operator.py

Док здесь: https://airflow.apache.org/_modules/airflow/operators/python_operator.html

0 голосов
/ 11 января 2019

Добавьте set -e; к вашему bash_command. Например: 

get_files = SSHExecuteOperator(
    task_id='get_files',
    bash_command='set -e; python example_script.py',
    ssh_hook=sshHook,
    dag=dag)
...