开发者社区问答正文

如何在Airflow中将参数传递给PythonOperator

我刚开始使用Airflow，任何人都可以告诉我如何将参数传递给PythonOperator，如下所示：

t5_send_notification = PythonOperator(

task_id='t5_send_notification',
provide_context=True,
python_callable=SendEmail,
op_kwargs=None,
#op_kwargs=(key1='value1', key2='value2'),
dag=dag,

)

def SendEmail(**kwargs):

msg = MIMEText("The pipeline for client1 is completed, please check.")
msg['Subject'] = "xxxx"
msg['From'] = "xxxx"
......
s = smtplib.SMTP('localhost')
s.send_message(msg)
s.quit()

我希望能够将一些参数传递给可t5_send_notification调用的SendEmail，理想情况下，我想将完整的日志和/或部分日志（实际上是从kwargs）附加到要发送的电子邮件中，猜测这t5_send_notification是收集这些信息的地方。

展开

收起

一码平川MACHEL 2019-02-28 11:30:45 12033 版权

1 条回答

写回答

取消提交回答

一码平川MACHEL

将dict对象传递给op_kwargs
使用键从python callable中的kwargs dict 访问它们的值

def SendEmail(**kwargs):

print(kwargs['key1'])
print(kwargs['key2'])
msg = MIMEText("The pipeline for client1 is completed, please check.")
msg['Subject'] = "xxxx"
msg['From'] = "xxxx"
......
s = smtplib.SMTP('localhost')
s.send_message(msg)
s.quit()

t5_send_notification = PythonOperator(

task_id='t5_send_notification',
provide_context=True,
python_callable=SendEmail,
op_kwargs={'key1': 'value1', 'key2': 'value2'},
dag=dag,

)

t5_send_notification = PythonOperator(

task_id='t5_send_notification',
provide_context=True,
python_callable=SendEmail,
op_kwargs={my_param='value1'},
dag=dag,

)

def SendEmail(my_param,**kwargs):

print(my_param) #'value_1'
msg = MIMEText("The pipeline for client1 is completed, please check.")
msg['Subject'] = "xxxx"
msg['From'] = "xxxx"
......
s = smtplib.SMTP('localhost')
s.send_me