python - 将 python 脚本转换为 Airflow PythonOperator

Question

我有一个可以运行的 python 脚本，可以从 CronJob 运行。我想使用 PythonOperator(s) 将其转换为 DAG，因为我们现在正在转换为 Airflow。

假设我有函数:a(),b(),c(),d()它们的执行顺序是:a->b->c->d

假设功能代码是:

def a(): 
    print("Happy")

def b(): 
    print("Birthday")

def c(): 
    print("to")

def d(): 
    print("you!")

** 这只是一个示例，我所有函数的代码都比较复杂

我有这个 DAG:

args = {
    'owner': 'airflow',
    'start_date': airflow.utils.dates.days_ago(2),
    'schedule_interval': '0 10 * * *'
}

dag = DAG(dag_id='example', default_args=args)

a = PythonOperator(task_id='a', dag=dag)
b = PythonOperator(task_id='b', dag=dag)
c = PythonOperator(task_id='c', dag=dag)
d = PythonOperator(task_id='d', dag=dag)

a.set_downstream(b)
b.set_downstream(c)
c.set_downstream(d)

我不明白的是我应该在哪里放置a(),b(),c(),d()的代码以及在执行时在哪里指定它们的名称Python 运算符。

你可以说我正在寻找一种将 Python 脚本转换为 Airflow 的方法，因为每个函数都是一个单独的运算符。

我认为这应该非常简单和基本，但我没有找到任何有关如何做到这一点的信息。

Answer 1

在python操作符中，将应该执行的python函数传入操作符中。因此，您需要传递一个 python_callable kwarg，如下所示:

def do_a():
    print('running a')

a = PythonOperator(task_id='a', python_callable=do_a, dag=dag)

运算符的源代码通常会记录它们的参数。 Python operator docs