Airflow REST API 간단히 사용해보기(with Python) - TIL230620

📚 KDT WEEK 12 DAY 2 TIL

• Airflow API 사용하기

 

---

사용하기 전에

 

먼저 Airflow API를 사용하기 위해 airflow.cfg 변경이 조금 필요하다.

 

airflow.cfg의 api 섹션에서 auth_backend의 값을 아래와 같이 변경해야한다.

 

[api]
auth_backend = airflow.api.auth.backend.basic_auth

 

basic_auth는 airflow api를 사용하기 위해 사용자의 아이디와 비밀번호로 인증하는 방식이다.

 

이 설정이 되어있는지 확인하고 싶다면 터미널에 다음의 명령어를 입력해서 확인할 수 있다. (docker)

 

docker exec -it airflow-scheduler airflow config get-value api auth_backend

airflow-scheduler에는 docker ps 명령어를 사용했을 때 나타난 airflow scheduler 이미지 이름을 넣으면 된다.

 

 

---

 

Airflow REST API

 

다음의 링크에서 자세한 정보를 얻을 수 있다.

 

Airflow API Documents

 

https://airflow.apache.org/docs/apache-airflow/stable/stable-rest-api-ref.html

Airflow REST API

 

 

Airflow API에서 통신할 때 사용하는 기본 포맷은 json이다.

 

Airflow REST API에서 명시한 부분

 

 

파이썬 코드를 통해 몇 가지 요청을 날려보자.

 

1. 모든 dags 리스트하기

 

# List dags

import requests
from requests.auth import HTTPBasicAuth
import pprint

url = "http://localhost:8080/api/v1/dags"

dags = requests.get(url, auth=HTTPBasicAuth("airflow","airflow"))

pprint.pprint(dags.json())

 

모든 dags 리스트 결과

 

dags 폴더안의 dag들이 모두 나타나게 된다.

is_active는 dags 폴더에 존재 유무에 따라 True, False로 나타나고,

is_paused는 dag 활성화 여부에 따라 True, False로 나타난다.

 

만약 is_paused = True인 dag만 나타내고 싶다면, 다음과 같은 조건을 넣으면 가능할 것이다.

 

for d in dags.json()["dags"]:
	if not d["is_paused"]:
    		print(d["dag_id"])

 

2. 모든 Variables 리스트하기

 

# List variables

url = "http://localhost:8080/api/v1/variables"

dags = requests.get(url, auth=HTTPBasicAuth("airflow","airflow"))

pprint.pprint(dags.json())

 

모든 variables 리스트 결과

 

dags를 리스트하는 과정과 비슷하다.

참고로 docker-compose에 작성된 환경변수는 리턴되지 않는다.

 

3. 모든 config 리스트하기

config의 경우 airflow.cfg에서 설정이 조금 필요하다.

 

airflow.cfg의 webserver > expose_config = True 로 설정해야한다.

 

docker-compose.yml의 environment에서 설정해줄 수도 있다.

 

environment:
    AIRFLOW__WEBSERVER__EXPOSE_CONFIG : 'true'

 

나는 후자를 선택하고, docker compose down > docker compose up 명령어로 재실행했다.

 

# List config

url = "http://localhost:8080/api/v1/config"

dags = requests.get(url, auth=HTTPBasicAuth("airflow","airflow"))

pprint.pprint(dags.text)

 

config가 잘 리스트된다.

 

4. 특정 DAG RUN 하기

 

# Run HelloWorld Dag

import json 

url = "http://localhost:8080/api/v1/dags/HelloWorld/dagRuns"
auth = HTTPBasicAuth("airflow", "airflow")
headers = {"Content-Type": "application/json"}
data = {"execution_date": "2023-05-24T00:00:00Z"}

response = requests.post(url, auth=auth, headers=headers, data=json.dumps(data))

response.json()

 

run의 경우 execution_date 전달이 필요하므로 post로 요청해야 한다.

execution_date에는 위와 같은 날짜 형식이어야한다.

 

정상적으로 동작했다.

 

5. Airflow 상태(Health) 확인하기

 

# Check Airlfow Health

url = "http://localhost:8080/api/v1/health"

dags = requests.get(url, auth=HTTPBasicAuth("airflow","airflow"))

pprint.pprint(dags.text)

 

정상적인 경우, healthy로 나타난다.