GCP Composer Airflow Dag에서 유용한 코드

 

목차

 

[ Composer Airflow 환경변수 ]

 

Composer는 GCP의 서비스 중 하나이다. Airflow인데 서비스를 Composer라고 지었고 내부는 Airflow 그 잡채이다. 환경변수를 사용해서 자주 사용하는 변수를 세팅할 수 있다. Composer 들어가서 "AIRFLOW UI 열기" 를 하면, 아래처럼 Airflow 화면을 만날 수 있다.

 

 

Airflow에서 "Admin > Variables" 로 이동하면 여러가지 세팅을 넣을 수 있다. 예를 들어서 "bucket" 이라는 키를 만들어서서 bucket url 을 설정하면 Dag Code를 작성할 때 불러와서 사용할 수 있다는 것이다. 매번 url 경로를 코드에 넣는게 번거로운 일인데 이렇게 환경변수를 설정하면 두고두고 사용할 수 있다.

 

 

파이썬으로 Dag 코드를 작성할 때 위 환경 변수를 사용하는 방법은 아래 코드와 같다. print를 진행해보면 환경변수에 넣어진 값을 잘 불러올 수 있다는 것을 알 수 있다.

from airflow import models

BUCKET = models.variable.Variable.get("bucket")
SPARK_CODE = BUCKET + "/dags/spark/code.py"

print(SPARK_CODE)

gs://asia-northeast3-#######-bucket/dag/spark/code.py

 

[ Composer Airflow 패키지 설치 ]

GCP내에서의 Airflow는 기본적으로 작업을 수행하는게 아니고 명령을 보내는 형태로 사용해야 한다. 그래야 작업에 지연이 없이 잘 동작이 된다. 만약 작업을 수행하더라고 많은 리소스를 사용하지 않는 작업에 한에서만 사용해야 한다.

 

예를 들어 할지 말아야할 작업은 아래와 같다.

• RDB에 쿼리를 조회해서 데이터를 가져오는 작업
• Dag안에서 데이터프레임을 전처리하는 행위
• Bucket에 있는 데이터프레임을 읽어오는 행위

그럼 지향해야하는 작업은 어떤 덧들이 있는지 알아보자.

• Bigquery가 작업을 수행하도록 쿼리를 BigQuery로 수행
• VM Instance가 Bash를 수행
• RDB 저장프로시저(Stored Procedure)를 실행

지향해야 하는 방법으로 Airflow를 사용하려면 필요한 패키지를 설치하곤 해야하는데 그 방법은 매우 간단하다.

"Composer > PYPI 패키지 > 수정" 으로 들어가 원하는 패키지의 이름을 넣고 버전을 "==" ">=" 등 과 같이 넣어서 설치하면 5분 정도 반영하는 시간이 소요되고 바로 사용이 가능하게 된다.