将python脚本转换为Airflow PythonOperator

本文介绍了将python脚本转换为Airflow PythonOperator的处理方法，对大家解决问题具有一定的参考价值，需要的朋友们下面随着小编来一起学习吧！ 问题描述

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

I have a working python script with runs from CronJob. I want to convert it to DAG with PythonOperator(s) as we now are converting to 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!")

**这只是一个例子，我所有函数的代码都更复杂

** This is just an example my code for all functions is more complex

我有这个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（），在执行PythonOperator的过程中应在何处指定其名称。

What I don't understand is where do I put the codes of a(),b(),c(),d() and where do I specify thier names in the execution of the PythonOperator.

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

You could say that I'm looking for a way to convert my Python script into Airflow as each function will be a separate operator.

我认为这应该非常简单和基本，但是我没有找到有关如何执行此操作的任何信息。

I thought this should be very simple and basic but I didn't find any information about how to do that.

推荐答案

在python运算符中，应执行的python函数传递到运算符中。因此，您将希望像这样传递 python_callable kwarg：

In the python operator, the python function that should be executed is passed into the operator. So you will want to pass a python_callable kwarg like so:

def do_a(): print('running a') a = PythonOperator(task_id='a', python_callable=do_a, dag=dag)

操作员的来源通常会为他们记录参数。 Python运算符文档

The source for the operators will usually document the params for them. Python operator docs