[Airflow] SimpleHttp 오퍼레이터 사용하기

1. SimpleHttpOperator

• HTTP 요청을 하고 결과로 text를 리턴 받는 오퍼레이터
  (리턴 값은 Xcom에 저장)
• HTTP를 이용하여 API를 처리하는 RestAPI 호출 시 사용가능
• 오퍼레이터 명세

 

 

 

 

2. Conn 등록 (서울시 공공 api 사용한 예제)

• 서울 열린데이터광장에 회원가입 후 api key를 발급 받음
• https://data.seoul.go.kr/

열린데이터광장 메인

 

• Admin -> Connections 에 들어가서 + 버튼 클릭

• Add Connection 작성

• Conn_id : openapi.seoul.go.kr (다른 conn 이름과 중복되지 않게 작성)
• Connection_type : HTTP
• Host : http://openapi.seoul.go.kr
• Port : 8088

 

2. api key 등록할 Variable 등록

• api key를 코드에 하드코딩 해도 동작하지만 보안 및 편의성을 위해 전역변수 Variable에 등록함

 

 

 

3. SimpleHttpOperator 사용 예제 코드

from airflow import DAG
from airflow.operators.bash import BashOperator
from airflow.providers.http.operators.http import SimpleHttpOperator
from airflow.decorators import task
import pendulum


with DAG(
    dag_id = 'dags_simple_http_operator',
    start_date = pendulum.datetime(2023, 4, 1, tz='Asia/Seoul'),
    catchup = False,
    schedule = None
) as dag:
    '''서울시 공공자전거 대여소 정보'''
    tb_cycle_station_info = SimpleHttpOperator(
        task_id = 'tb_cycle_station_info',
        http_conn_id = 'openapi.seoul.go.kr',
        endpoint = '{{var.value.apikey_openapi_seoul_go_kr}}/json/tbCycleStationInfo/1/10/',
        method = 'GET',
        headers = {
            'Content-Type' : 'application/json',
            'charset' : 'utf-8',
            'Accept' : '*/*'
        }
    )

    @task(task_id='python_2')
    def python_2(**kwargs):
        ti = kwargs['ti']
        rslt = ti.xcom_pull(task_ids='tb_cycle_station_info')

        import json
        from pprint import pprint

        pprint(json.loads(rslt))

    tb_cycle_station_info >> python_2()

 

실행결과 Graph

 

실행결과 XCom

 

 

실행결과 Log