Как передать параметр в PythonOperator в Airflow - PullRequest
Новая
       9

Как передать параметр в PythonOperator в Airflow

0 голосов
/ 27 февраля 2019

Я только начал использовать Воздушный поток , может кто-нибудь просветить меня, как передать параметр в 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 - это место для сбора этой информации.

Большое спасибо.

Ответы [ 2 ]

0 голосов
/ 27 февраля 2019
  1. Передайте объект dict op_kwargs

  2. Используйте ключи для доступа к их значению из kwargs dict в вашем вызываемом Python 

    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,
)
0 голосов
/ 27 февраля 2019

Это должно работать: 

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
...