Поток воздуха не прерывается при ошибке подзадачи - PullRequest
Новая
       9

Поток воздуха не прерывается при ошибке подзадачи

0 голосов
/ 30 мая 2018

Возможно, кто-то может сказать мне, что я здесь делаю неправильно.У меня есть задача в Airflow, которая запускает команду, и в логах я получаю эту ошибку: 

[2018-05-30 11:22:43,814] {models.py:1428} INFO - Executing <Task(PythonOperator): computer_unload_and_load> on 2018-05-30 15:22:41.595535
[2018-05-30 11:22:43,814] {base_task_runner.py:115} INFO - Running: ['bash', '-c', 'airflow run copy_kiosk_status computer_unload_and_load 2018-05-30T15:22:41.595535 --job_id 23 --raw -sd DAGS_FOLDER/copy_poc.py']
[2018-05-30 11:22:44,367] {base_task_runner.py:98} INFO - Subtask: [2018-05-30 11:22:44,367] {__init__.py:45} INFO - Using executor SequentialExecutor
[2018-05-30 11:22:44,412] {base_task_runner.py:98} INFO - Subtask: [2018-05-30 11:22:44,412] {models.py:189} INFO - Filling up the DagBag from /other/airflow/dags/copy_kiosk.py
[2018-05-30 11:22:44,570] {cli.py:374} INFO - Running on host [redacted]
[2018-05-30 11:22:46,967] {base_task_runner.py:98} INFO - Subtask: ERROR:  schema "dw2" does not exist
[2018-05-30 11:22:46,969] {base_task_runner.py:98} INFO - Subtask: [2018-05-30 11:22:46,968] {python_operator.py:90} INFO - Done. Returned value was: None

Вот задача: 

computer_load_task = PythonOperator(
    task_id='computer_unload_and_load',
    python_callable=unload_and_load.unload_and_load_table,
    op_args=(source_host, source_db, "public", "computer", "computer_unload_task")
    dag=dag
)

Вот функция, которую она вызывает: 

def load_table(host, db, schema, table, unload_task_id=False, file_path=False, **kwargs):
    """ load a csv file into a table
        if no file_path is given, uses XCOM to get the file_name returned by the unload task"""
    try:
        if not file_path:
            file_path = kwargs['ti'].xcom_pull(task_ids=unload_task_id)

        load_cmd = "\copy {}.{} FROM {} WITH (FORMAT CSV,  NULL '^',  HEADER)".format(schema, table, file_path)
        command = [ "psql",
                    "-U", "root",
                    "-h", host,
                    "-d", db,
                    "-c", load_cmd
                  ]
        subprocess.run(command)

    except:
        raise

Очевидно, я знаю, как исправить ошибку (там у меня неверная схема), но я хочу знать, почему задачи выполняются в Airflow, а не при сбое?Должно быть, я упускаю что-то очевидное.

1 Ответ

0 голосов
/ 31 мая 2018

Ваш пример кода выглядит отлично - хотя бы одна вещь.В этом примере кода provide_context=True отсутствует в вашей задаче.

Кроме того, я думаю это связано с ошибками обработки подпроцесса.Вы можете попробовать установить атрибут check=True, чтобы подпроцесс вызывал исключение, если что-то пойдет не так.

В качестве альтернативы вы можете проверить подпроцесс check_returncode() и выдать собственное исключение, если оно ненулевое.

Вот документы Python 3 для подпроцесса: https://docs.python.org/3/library/subprocess.html

Чтобы отдельно попробовать обработку исключений, было бы неплохо создать задачу и просто raise исключение, если это не сработало.

...