[Airflow] python 오퍼레이터 op_args & op_kwagrs

1. Python 오퍼레이터 op_args 사용법

case1) 함수에 일반 변수만 있을 경우

def regist(name, sex):
	print(f'이름은 {name}이고 성별은 {sex}입니다')
    
python_task = PythonOperator(
	task_id='python_task',
    python_callable=regist,
    op_args=['hjkim','man']  # 리스트로 작성
)

 

 

case2) 함수에 일반 변수 + *args도 있을 경우

def regist(name, sex, *args):
	print(name)
    print(sex)
    print(args)  # ('kr','seoul') 형태로 나옴
    
python_task = PythonOperator(
	task_id='python_task',
    python_callable=regist,
    op_args=['hjkim','man','kr','seoul]  # 리스트로 작성
)

 

 

case3) 함수에 *args 변수만 있을 경우

def regist(*args):
    print(args)  # ('hjkim','man','kr','seoul') 형태로 나옴
    
python_task = PythonOperator(
	task_id='python_task',
    python_callable=regist,
    op_args=['hjkim','man','kr','seoul]  # 리스트로 작성
)

 

 

 

2. Python 오퍼레이터 op_args  사용 예제코드

"""
common_func.py
"""

def regist(name, sex, *args):
    print(f'이름: {name}')
    print(f'성별: {sex}')
    print(f'기타옵션들: {args}')

 

 

"""
dags_python_with_op_args.py
"""

from airflow import DAG
import pendulum
import datetime
from airflow.operators.python import PythonOperator
from common.common_func import regist
with DAG(
    dag_id="dags_python_with_op_args",
    schedule="30 6 * * *",
    start_date=pendulum.datetime(2023, 3, 1, tz="Asia/Seoul"),
    catchup=False
) as dag:
    regist_t1 = PythonOperator(
        task_id='regitst_t1',
        python_callable=regist,
        op_args=['hjkim','man','kr','seoul']
    )

    regist_t1

 

실행결과

 

 

 

 

 

3. Python 오퍼레이터 op_kwargs 사용법

case1) 함수에 일반 변수만 있을 경우

def regist(name, sex):
	print(f'이름은 {name}이고 성별은 {sex}입니다')
    
python_task = PythonOperator(
	task_id='python_task',
    python_callable=regist,
    op_kwargs={'name':'hjkim','sex':'man'}  # 딕셔너리로 작성
)

 

case2) 함수에 일반 변수 + **kwargs도 있을 경우

def regist(name, sex, **kwargs):
	print(name)
    print(sex)
    print(kwargs)
    
python_task = PythonOperator(
	task_id='python_task',
    python_callable=regist,
    op_kwargs={
    	'name':'hjkim',
        'sex':'man',
        'country':'kr',
        'city':'seoul'
    } # 딕셔너리로 작성
)

 

case3) 함수에 **kwargs 변수만 있을 경우

def regist(**kwargs):
	name=kwargs['name'] or ''
    sex=kwargs['sex'] or ''
    country=kwargs['country'] or ''
    city = kwargs['city'] or ''
    print(f'name은 {name}이고,
    	성별은 {sex}이고,
        국가는 {country}이고,
        도시는 {city} 입니다.')
    
python_task = PythonOperator(
	task_id='python_task',
    python_callable=regist,
    op_kwargs={
    	'name':'hjkim',
        'sex':'man',
        'country':'kr',
        'city':'seoul'
    } # 딕셔너리로 작성
)

 

 

case4) 함수에 파라미터 + *args + **kwargs 모두 있는 경우

def regist(name, sex, *args, **kwargs):
	print(name)
    print(sex)
    print(args)
    print(kwargs)
    
    
python_task_2 = PythonOperator(
	task_id='python_task_2',
    python_callable=regist,
    op_args=['hjkim','man','kr','seoul'],
    op_kwargs={'phone':010, 'email':'fdew@naver.com'}
)

 

 

 

2. Python 오퍼레이터 op_kwargs 사용 예제코드

"""
common_func.py
"""

def regist2(name, sex, *args, **kwargs):
    print(f'이름: {name}')
    print(f'성별: {sex}')
    print(f'기타옵션들: {args}')
    email = kwargs['email'] or None
    phone = kwargs['phone'] or None

    if email:
        print(email)
    if phone:
        print(phone)

 

"""
dags_python_with_op_kwargs.py
"""

from airflow import DAG
import pendulum
import datetime
from airflow.operators.python import PythonOperator
from common.common_func import regist2
with DAG(
    dag_id="dags_python_with_op_kwargs",
    schedule="30 6 * * *",
    start_date=pendulum.datetime(2023, 3, 1, tz="Asia/Seoul"),
    catchup=False
) as dag:
    regist2_t1 = PythonOperator(
        task_id='regitst2_t1',
        python_callable=regist2,
        op_args=['hjkim','man','kr','seoul'],
        op_kwargs={'email':'fdew@naver.com','phone':'010'}
    )

    regist2_t1

 

 

실행결과