Cloud Composer
- 파이프라인을 작성하여 예약 및 모니터링 할 수 있는 통합 워크플로 관리 서비스
- Apache Airflow based
- 구글 쿠버네티스 엔진 : 배포환경
- Cloud SQL : 메타데이터 저장
- App Engine : Airflow web server hosting
- Stackdriver : 로그 관리
- Pyhton을 기반으로 DAG와 Task에 대한 코드를 작성 가능
- 분산 환경 및 웹 UI 기반의 강력한 모니터링 기능을 제공
Apache Airflow
- Airbnb에서 개발된 워크플로 통합 도구
- Python을 기본으로 태스크에 대한 코드를 작성할 수 있음
- 웹 UI 기반의 강력한 모니터링 룰
Cloud Composer 주요 개념
- 워크 플로우는 Directed Acyclic Graph (DAG)로 표현 됨
DAG
my_dag = DAG(
'may_dag',
description='My First DAG',
start_date=datetime(2022,8, 30),
schedule_interval='1 0 * * 2',
catchup=False
)
- Operator : DAG 안에 정의되는 작업 함수, 이를 이용하여 Task를 작성
Bash Operator
task_echo_hello_world = BashOperator(
task_id='echo_hello_world',
bash_command='echo "Hello World (shell)"',
dag=my_dag
)
Python Operator
def print_hello_world():
print("Hello World!!! (python)")
task_print_hello_world = PythonOperator(
task_id='print_hello_world',
python_callable=print_hello_world,
dag=my_dag
)
데이터 저장
- Cloud Composer 생성 시, GCS 내에 버킷을 만들고 클라우드 스토리지 FUSE를 사용하여 에어플로 인스턴스와 GCS 버킷을 서로 매핑
| 폴더 | GCS 경로 | 매핑된 디렉터리 | 용량 제한 |
|---|---|---|---|
| DAG | gs://bucket-name/dags | /home/airflow/gcs/dags | 기본적으로 100GB용량 제공 |
| Plugins | gs://bucket-name/plugins | /home/airflow/gcs/plugins | 기본적으로 100GB용량 제공 |
| Data | gs://bucket-name/data | /home/airflow/gcs/data | 용량 제한 없음 |
| Logs | gs://bucket-name/logs | /home/airflow/gcs/logs | 용량 제한 없음 |