python - Airflow，我如何使用 BashOperator 通过 python name.py 运行 .py 文件

Question

我正在使用 Celery 和 Redis 运行 Airflow，它工作得很好，但我在工作端遇到了问题。我有两个 docker-compose 文件，一个是在服务器上运行的 master，一个是在其他 pc 上运行的 worker。

我有运行 python script.py 的 dag，但总是失败，因为它找不到脚本。似乎 Airflow Base 任务运行器只是将 dag 文件复制到 tmp 文件夹。

my dags folder is like:
    dags/ 
        test_dag.py
        test.py

test_dag.py

from airflow.models import DAG
from airflow.operators.bash_operator import BashOperator
from datetime import datetime, timedelta
default_args = {
        'owner': 'airflow',
        'start_date': datetime(2018, 9, 3)
        }
dag = DAG('test', default_args=default_args, schedule_interval='*/20 * * * *', catchup=False)

curl = BashOperator(
    task_id='testingbash',
    bash_command="python test.py",
    dag=dag)

测试.py

print('it worked')

错误:

[2018-09-06 22:30:34,832] {base_task_runner.py:107} INFO - Job 4: Subtask testingbash [2018-09-06 22:30:34,832] {cli.py:492} INFO - Running <TaskInstance: test.testingbash 2018-09-06T22:00:00+00:00 [running]> on host e91df5a905de
[2018-09-06 22:30:39,066] {bash_operator.py:74} INFO - Tmp dir root location: 
 /tmp
[2018-09-06 22:30:39,067] {bash_operator.py:87} INFO - Temporary script location: /tmp/airflowtmprgxtunoa/testingbash_7opnm28
[2018-09-06 22:30:39,067] {bash_operator.py:97} INFO - Running command: python test.py
[2018-09-06 22:30:39,079] {bash_operator.py:106} INFO - Output:
[2018-09-06 22:30:39,164] {bash_operator.py:110} INFO - python: can't open file 'test.py': [Errno 2] No such file or directory
[2018-09-06 22:30:39,165] {bash_operator.py:114} INFO - Command exited with return code 2
[2018-09-06 22:30:40,400] {models.py:1736} ERROR - Bash command failed
Traceback (most recent call last):
  File "/usr/local/lib/python3.6/site-packages/airflow/models.py", line 1633, in _run_raw_task
    result = task_copy.execute(context=context)
  File "/usr/local/lib/python3.6/site-packages/airflow/operators/bash_operator.py", line 118, in execute
    raise AirflowException("Bash command failed")
airflow.exceptions.AirflowException: Bash command failed
[2018-09-06 22:30:40,408] {models.py:1764} INFO - Marking task as FAILED.
[2018-09-06 22:30:42,132] {base_task_runner.py:107} INFO - Job 4: Subtask testingbash Traceback (most recent call last):
[2018-09-06 22:30:42,132] {base_task_runner.py:107} INFO - Job 4: Subtask testingbash   File "/usr/local/bin/airflow", line 32, in <module>
[2018-09-06 22:30:42,132] {base_task_runner.py:107} INFO - Job 4: Subtask testingbash     args.func(args)
[2018-09-06 22:30:42,133] {base_task_runner.py:107} INFO - Job 4: Subtask testingbash   File "/usr/local/lib/python3.6/site-packages/airflow/utils/cli.py", line 74, in wrapper
[2018-09-06 22:30:42,133] {base_task_runner.py:107} INFO - Job 4: Subtask testingbash     return f(*args, **kwargs)
[2018-09-06 22:30:42,133] {base_task_runner.py:107} INFO - Job 4: Subtask testingbash   File "/usr/local/lib/python3.6/site-packages/airflow/bin/cli.py", line 498, in run
[2018-09-06 22:30:42,133] {base_task_runner.py:107} INFO - Job 4: Subtask testingbash     _run(args, dag, ti)
[2018-09-06 22:30:42,133] {base_task_runner.py:107} INFO - Job 4: Subtask testingbash   File "/usr/local/lib/python3.6/site-packages/airflow/bin/cli.py", line 402, in _run
[2018-09-06 22:30:42,134] {base_task_runner.py:107} INFO - Job 4: Subtask testingbash     pool=args.pool,
[2018-09-06 22:30:42,134] {base_task_runner.py:107} INFO - Job 4: Subtask testingbash   File "/usr/local/lib/python3.6/site-packages/airflow/utils/db.py", line 74, in wrapper
[2018-09-06 22:30:42,134] {base_task_runner.py:107} INFO - Job 4: Subtask testingbash     return func(*args, **kwargs)
[2018-09-06 22:30:42,134] {base_task_runner.py:107} INFO - Job 4: Subtask testingbash   File "/usr/local/lib/python3.6/site-packages/airflow/models.py", line 1633, in _run_raw_task
[2018-09-06 22:30:42,134] {base_task_runner.py:107} INFO - Job 4: Subtask testingbash     result = task_copy.execute(context=context)
[2018-09-06 22:30:42,134] {base_task_runner.py:107} INFO - Job 4: Subtask testingbash   File "/usr/local/lib/python3.6/site-packages/airflow/operators/bash_operator.py", line 118, in execute
[2018-09-06 22:30:42,134] {base_task_runner.py:107} INFO - Job 4: Subtask testingbash     raise AirflowException("Bash command failed")
[2018-09-06 22:30:42,135] {base_task_runner.py:107} INFO - Job 4: Subtask testingbash airflow.exceptions.AirflowException: Bash command failed
[2018-09-06 22:30:46,945] {logging_mixin.py:95} INFO - [2018-09-06 22:30:46,944] {jobs.py:2612} INFO - Task exited with return code 1

解决方法:

几天前修复了，我没有考虑我的 docker 正在构建的结构，现在我使用 python ~/scripts/test.py 运行并且运行良好。

Answer 1

如果您只想运行 python 脚本，使用 PythonOperator 可能更容易。

如果您打算使用 BashOperator，您只需要包含文件的绝对文件路径 - 默认情况下，它会创建并查找 tmp directory.