python - 如何使用 AirFlow 运行 python 文件的文件夹？

Question

我在 python 文件的文件夹中有一系列 Python 任务：file1.py，file2.py，...

我阅读了 Airflow 文档，但没有看到如何在 DAG 中指定 python 文件的文件夹和文件名？

我想执行那些 python 文件（不是通过 Python Operator 的 Python 函数）。

任务1：执行file1.py（带有一些导入包）

任务2：执行file2.py（与其他一些导入包）

这会很有帮助。感谢和问候

Answer 1

我知道你在问你“想要执行那些 python 文件（而不是通过 Python Operator 的 Python 函数）”。但我认为这可能比您使用 Airflow 的效率低。我在之前写的答案中也看到了混乱，所以这是您想要的方式，也是我建议的完成任务的方式：

假设：

dags/
    my_dag_for_task_1_and_2.py
    tasks/
         file1.py
         file2.py

您要求避免PythonOperator：

#  my_dag_for_task_1_and_2.py
import datetime as dt
from airflow import DAG
from airflow.operators import BashOperator

with DAG(
    'my_dag_for_task_1_and_2',
    default_args={
        'owner': 'me',
        'start_date': datetime(…),
        …,
    }, 
    schedule_interval='8 * * * *',
) as dag:
    task_1 = BashOperator(
        task_id='task_1', 
        bash_command='/path/to/python /path/to/dags/tasks/file1.py',
    )
    task_2 = BashOperator(
        task_id='task_2', 
        bash_command='/path/to/python /path/to/dags/tasks/file2.py',
    )
    task_1 >> task_2

您没有从头开始为 Airflow 编写 Python，而是使用PythonOperator：

#  my_dag_for_task_1_and_2.py
import datetime as dt
from airflow import DAG
from airflow.operators import PythonOperator
import tasks.file1
import tasks.file2

with DAG(
    'my_dag_for_task_1_and_2',
    default_args={
        'owner': 'me',
        'start_date': datetime(…),
        …,
    }, 
    schedule_interval='8 * * * *',
) as dag:
    task_1 = PythonOperator(
        task_id='task_1', 
        python_callable=file1.function_in_file1,
    )
    task_2 = PythonOperator(
        task_id='task_2', 
        python_callable=file2.function_in_file2,  # maybe main?
    )
    task_1 >> task_2

Answer 2

要执行整个 python 文件，请使用BashOperator（如 liferacer 的答案）：

from airflow.operators.bash_operator import BashOperator

bash_task = BashOperator(
    task_id='bash_task',
    bash_command='python file1.py',
    dag=dag
)

PythonOperator然后，使用调用你的main函数来做到这一点。您应该已经有一个__main__块，所以将其中发生的事情放入一个main函数中，这样您的file1.py看起来像这样：

def main():
    """This gets executed if `python file1` gets called."""
    # my code

if __name__ == '__main__':
    main() 

然后你的 dag 定义：

from airflow.operators.python_operator import PythonOperator

import file1

python_task = PythonOperator(
    task_id='python_task',
    python_callable=file1.main,
    dag=dag
)

Answer 3

您可以使用 BashOperator 将 python 文件作为任务执行

    from airflow import DAG
    from airflow.operators import BashOperator,PythonOperator
    from datetime import datetime, timedelta

    seven_days_ago = datetime.combine(datetime.today() - timedelta(7),
                                      datetime.min.time())

    default_args = {
        'owner': 'airflow',
        'depends_on_past': False,
        'start_date': seven_days_ago,
        'email': ['airflow@airflow.com'],
        'email_on_failure': False,
        'email_on_retry': False,
        'retries': 1,
        'retry_delay': timedelta(minutes=5),
      )

    dag = DAG('simple', default_args=default_args)
t1 = BashOperator(
    task_id='testairflow',
    bash_command='python /home/airflow/airflow/dags/scripts/file1.py',
    dag=dag)