[Airflow] Custom 오퍼레이터 (1)

 

 

1. Airflow의 꽃, Custom 오퍼레이터

• Airflow는 오퍼레이터를 직접 만들어 사용할 수 있도록 클래스를 제공함
  (BaseOperator)
• BaseOperator 상속시 두 가지 메서드를 재정의해야 함 (Overriding)
  (1) def __init__
  - 클래스에서 객체 생성 시 객체에 대한 초기값을 지정하는 함수
  (2) def execute(self, context)
  - __init__ 생성자로 객체를 얻은 후 execute 메서드를 실행시키도록 되어있음
  - 비즈니스 로직은 execute에 구현 필요함
• https://airflow.apache.org/docs/apache-airflow/stable/howto/custom-operator.html

 

 

2. Custom 오퍼레이터 개발 Study

• 오퍼레이터 기능 정의
  - 서울시 공공데이터 api를 호출하여 전체 데이터를 받은 후 .csv 파일로 저장하기
• 오퍼레이터와 dag의 위치

경로	파일명	내용
/dags	dags_seoul_api.py	from operators.seoul_api_to_csv_operator import SeoulApiToCsvOperator
/plugins/operators	seoul_api_to_csv_operator.py	(만들 대상) class SeoulApiToCsvOperator

 

• Template 사용 가능한 파라미터 지정하기

 

3. Custom 오퍼레이터 기본 예제코드

'''
hello_operator.py
'''

from airflow.models.baseoperator import BaseOperator


class HelloOperator(BaseOperator):

	# 파라미터 중 어떤 파라미터에 템플레이팅 문법을 적용할 수 있게 할지 정의
	template_fields: Sequence[str] = ("name",)

	# 1. 생성자
    ## name 변수가 오퍼레이터를 통해 개체를 만들 때 입력하는 파라미터 값임
    ## ex.
    ## hello_task = HelloOperator(
    ##	task_id = 'xxxx'
    ##	name = '~~~~'   <- 이부분
    ## )
    def __init__(self, name: str, world: str, **kwargs) -> None:
    	# 부모함수(BaseOperator)의 생성자를 호출하는 부분
        # **kwargs에는 생성자에 명시되지 않은 파라미터들이 있음
        super().__init__(**kwargs)
        self.name = name      # 템플레이팅 문법적용 O
        self.world = world    # 템플레이팅 문법적용 x

	# 2. 실행하는 함수
    def execute(self, context):
        message = f"Hello {self.name}"
        print(message)
        return message