How to pass parameter to PythonOperator in Airflow

Question

I just started using Airflow, can anyone enlighten me how to pass a parameter into PythonOperator like below:

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()

I would like to be able to pass some parameters into the t5_send_notification's callable which is SendEmail, ideally I want to attach the full log and/or part of the log (which is essentially from the kwargs) to the email to be sent out, guessing the t5_send_notification is the place to gather those information.

Thank you very much.

Answer 1

1. 将dict对象传递给op_kwargs

2. 使用键从可调用的python中的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,
   )
   

Answer 2

PythonOperator有一个命名参数op_kwargs并接受dict对象。

有

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