Блог

На моем компьютере с Windows 10 установлен Airflow 1.10.11 в WSL 2 (Ubuntu-20.04).

У меня есть задача BashOperator, которая вызывает .EXE в Windows (через /mnt/c / … или через символическую ссылку). Задача завершается с ошибкой. Журнал показывает:

 [2020-12-16 18:34:11,833] {bash_operator.py:134} INFO - Temporary script location: /tmp/airflowtmp2gz6d79p/download.legacyFilesnihvszli
[2020-12-16 18:34:11,833] {bash_operator.py:146} INFO - Running command: /mnt/c/Windows/py.exe
[2020-12-16 18:34:11,836] {bash_operator.py:153} INFO - Output:
[2020-12-16 18:34:11,840] {bash_operator.py:159} INFO - Command exited with return code 1
[2020-12-16 18:34:11,843] {taskinstance.py:1150} ERROR - Bash command failed
Traceback (most recent call last):
  File "/usr/local/lib/python3.8/dist-packages/airflow/models/taskinstance.py", line 984, in _run_raw_task
    result = task_copy.execute(context=context)
  File "/usr/local/lib/python3.8/dist-packages/airflow/operators/bash_operator.py", line 165, in execute
    raise AirflowException("Bash command failed")
airflow.exceptions.AirflowException: Bash command failed
[2020-12-16 18:34:11,844] {taskinstance.py:1187} INFO - Marking task as FAILED. dag_id=test-dag, task_id=download.files, execution_date=20201216T043701, start_date=20201216T073411, end_date=20201216T073411
 

И это все. Возвращает код 1 без дополнительной полезной информации.

Запуск того же самого EXE-файла через bash работает отлично, без ошибок (я также пробовал это в своей собственной программе, которая что-то выдает на консоль — в bash он выдает просто отлично, но через airflow scheduler ту же ошибку 1).

Еще несколько данных и вещей, которые я сделал, чтобы исключить любую другую проблему:

• airflow scheduler запускается от имени root. Я также подтвердил, что он работает в root контексте, введя whoami команду в мой BashOperator, которая действительно была отправлена root (я должен также отметить, что все собственные программы Linux работают просто отлично! только программы Windows этого не делают.)
• Windows EXE, который я пытаюсь выполнить, и его каталог имеют полные разрешения «Все» (в моей собственной программе, конечно, я бы не осмелился сделать это в моей папке Windows — это был просто пример.)
• Сбой происходит как при доступе через /mnt / c, так и через символическую ссылку. В случае символической ссылки символическая ссылка имеет 777 разрешений.
• Я попытался запустить airflow test задачу BashOperator — она работает отлично — выдает вывод на консоль и возвращает 0 (успех).
• Пробовал с различными EXE-файлами — как «родными» (например, с Windows), так и с моими программами, созданными на C #. Одинаковое поведение во всех.
• Не найдено ни одной подобной проблемы, задокументированной в репозитории GitHub от Airflow, ни здесь, в Stack Overflow.

Вопрос в том, как использование Python в Airflow для подпроцесса (который планировщик воздушного потока использует для запуска операторов Bash) отличается от «обычного» Bash, вызывая error 1 ?

вы можете использовать подпроцесс библиотеки и систему Python и PowerShell

В папке Airflow> Dag создайте 2 файла: main.py и caller.py

итак, main.py позвоните caller.py и caller.py зайдите в компьютер (Windows), чтобы запустить файлы или процедуры.

Это процесс:

код Main.py:

 # Importing the libraries we are going to use in this example
from airflow import DAG
from datetime import datetime, timedelta
from airflow.operators.bash_operator import BashOperator


# Defining some basic arguments
default_args = {
   'owner': 'your_name_here',
   'depends_on_past': False,
   'start_date': datetime(2019, 1, 1),
   'retries': 0,
   }


# Naming the DAG and defining when it will run (you can also use arguments in Crontab if you want the DAG to run for example every day at 8 am)
with DAG(
       'Main',
       schedule_interval=timedelta(minutes=1),
       catchup=False,
       default_args=default_args
       ) as dag:

# Defining the tasks that the DAG will perform, in this case the execution of two Python programs, calling their execution by bash commands
    t1 = BashOperator(
       task_id='caller',
       bash_command="""
       cd /home/[Your_Users_Name]/airflow/dags/
       python3 Caller.py
       """)

    # copy t1, paste, rename t1 to t2 and call file.py
    
# Defining the execution pattern
    t1

    # comment: t1 execute and call t2
    # t1 >> t2
 

Код Caller.py

 import subprocess, sys

p = subprocess.Popen(["powershell.exe"
                     ,"cd C:\Users\[Your_Users_Name]\Desktop; python file.py"] # file .py
                    #,"cd C:\Users\[Your_Users_Name]\Desktop; .file.html"]    # file .html
                    #,"cd C:\Users\[Your_Users_Name]\Desktop; .file.bat"]     # file .bat
                    #,"cd C:\Users\[Your_Users_Name]\Desktop; .file.exe"]     # file .exe
                    , stdout=sys.stdout
                     )

p.communicate()
 

Как узнать, будет ли ваш код работать в airflow, если он запущен, все в порядке.