Apache Airflow (十二) ：PythonOperator
🏡 个人主页：IT贫道_大数据OLAP体系技术栈,Apache Doris,Clickhouse 技术-CSDN博客
🚩 私聊博主：加入大数据技术讨论群聊，获取更多大数据资料。
🔔 博主个人B栈地址：豹哥教你大数据的个人空间-豹哥教你大数据个人主页-哔哩哔哩视频
PythonOperator可以调用Python函数，由于Python基本可以调用任何类型的任务，如果实在找不到合适的Operator，将任务转为Python函数，使用PythonOperator即可。
关于PythonOperator常用参数如下，更多参数可以查看官网：airflow.operators.python — Airflow Documentation
python_callable(python callable)：调用的python函数
op_kwargs(dict)：调用python函数对应的 **args 参数，dict格式，使用参照案例。
op_args(list)：调用python函数对应的 *args 参数，多个封装到一个tuple中，list格式，使用参照案例。PythonOperator调度案例
import random
from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.python import PythonOperator
# python中 * 关键字参数允许你传入0个或任意个参数，这些可变参数在函数调用时自动组装为一个tuple。
# python中 ** 关键字参数允许你传入0个或任意个含参数名的参数,这些关键字参数在函数内部自动组装为一个dict。
def print__hello1(*a,**b):
print(a)
print(b)
print("hello airflow1")
# 返回的值只会打印到日志中
return{"sss1":"xxx1"}
def print__hello2(random_base):
print(random_base)
print("hello airflow2")
# 返回的值只会打印到日志中
return{"sss2":"xxx2"}
default_args = {
'owner':'maliu',
'start_date':datetime(2021, 10, 1),
'retries': 1, # 失败重试次数
'retry_delay': timedelta(minutes=5) # 失败重试间隔
}
dag = DAG(
dag_id = 'execute_pythoncode',
default_args=default_args,
schedule_interval=timedelta(minutes=1)
)
first=PythonOperator(
task_id='first',
#填写 print__hello1 方法时，不要加上“()”
python_callable=print__hello1,
# op_args 对应 print_hello1 方法中的a参数
op_args=[1,2,3,"hello","world"],
# op_kwargs 对应 print__hello1 方法中的b参数
op_kwargs={"id":"1","name":"zs","age":18},
dag = dag
)
second=PythonOperator(
task_id='second',
#填写 print__hello2 方法时，不要加上“()”
python_callable=print__hello2,
# random_base 参数对应 print_hello2 方法中参数“random_base”
op_kwargs={"random_base":random.randint(0,9)},
dag=dag
)
first >> second