Cloud Composer 1 | Cloud Composer 2 | Cloud Composer 3
이 페이지에서는 Cloud Composer가 Cloud Storage에 저장하는 사용자 환경 데이터를 설명합니다.
사용자가 환경을 만들면 Cloud Composer가 Cloud Storage 버킷을 만들고 사용자 환경과 버킷을 연결합니다. 버킷 이름은 환경 리전, 이름, 임의의 ID를 기반으로 합니다(예: us-central1-b1-6efabcde-bucket).
Cloud Composer는 환경 버킷의 특정 폴더를 환경에서 실행되는 Airflow 구성요소에 동기화합니다. 예를 들어 환경의 버킷에서 Airflow DAG 코드가 포함된 파일을 업데이트하면 Airflow 구성요소에도 업데이트된 버전이 전송됩니다. Cloud Composer는 동기화에 Cloud Storage FUSE를 사용합니다.
Cloud Storage 버킷의 폴더
| 폴더 | 스토리지 경로 | 매핑되는 디렉토리 | 설명 |
|---|---|---|---|
| DAG | gs://bucket-name/dags |
/home/airflow/gcs/dags |
사용자 환경의 DAG를 저장합니다. |
| 플러그인 | gs://bucket-name/plugins |
/home/airflow/gcs/plugins |
커스텀 사내 Airflow 연산자, 후크, 센서 또는 인터페이스와 같은 커스텀 플러그인을 저장합니다. |
| 데이터 | gs://bucket-name/data |
/home/airflow/gcs/data |
태스크가 생성하고 사용하는 데이터를 저장합니다. |
| 로그 | gs://bucket-name/logs |
/home/airflow/gcs/logs |
태스크의 Airflow 로그를 저장합니다. Airflow 웹 인터페이스 및 Google Cloud 콘솔의 로그 탭에서도 로그를 사용할 수 있습니다. |
Cloud Composer는 dags/ 폴더와 plugins/ 폴더를 단방향으로 동기화합니다. 단방향 동기화는 Airflow 구성요소의 로컬 변경사항을 이러한 폴더에 덮어쓴다는 것을 의미합니다. data/ 폴더와 logs/ 폴더는 양방향으로 동기화됩니다.
데이터 동기화는 eventual consistency를 갖습니다. 한 연산자에서 다른 연산자로 메시지를 보내려면 XComs를 사용하세요.
용량 고려 사항
dags/, plugins/, data/ 폴더의 데이터가 Airflow 스케줄러 및 작업자에 동기화됩니다.
plugins/ 폴더의 콘텐츠가 Airflow 웹 서버에 동기화됩니다.
데이터가 많이 입력되면 폴더가 Airflow 구성요소의 로컬 스토리지에서 더 많은 공간을 차지합니다. dags/ 및 plugins/에 데이터를 너무 많이 저장하면 작업이 중단되고 다음과 같은 문제가 발생할 수 있습니다.
작업자 또는 스케줄러가 로컬 스토리지 부족을 겪게 되고 구성요소의 로컬 디스크에서 부족한 공간으로 인해 삭제됩니다.
dags/폴더 및plugins/폴더에서 작업자 및 스케줄러로 파일을 동기화하는 데 시간이 오래 걸립니다.dags/및plugins/폴더의 파일을 작업자와 스케줄러에 동기화할 수 없습니다. 예를 들어dags/폴더에 2GB 파일을 저장하지만 Airflow 작업자의 로컬 디스크는 1GB만 수용할 수 있습니다. 동기화하는 동안 작업자의 로컬 스토리지가 부족하면 동기화를 완료할 수 없습니다.
DAG 및 플러그인 폴더
DAG 실행 오류를 방지하기 위해 Python 모듈에 DAG 또는 플러그인이 없더라도 DAG, 플러그인, Python 모듈을 dags/ 폴더 또는 plugins/ 폴더에 저장합니다.
예를 들어 py_file Dataflow 파이프라인을 참조하는 DataFlowPythonOperator를 사용합니다. py_file에는 DAG 또는 플러그인이 포함되지 않지만 dags/ 또는 plugins/ 폴더에 저장해야 합니다.
데이터 폴더
data/ 폴더의 특정 파일이 특정 Airflow 구성요소에 동기화되는 시나리오가 있습니다. 예를 들어 다음을 수행하는 중에 Cloud Composer가 특정 파일을 처음으로 읽기 시도하는 경우입니다.
DAG 파싱: DAG 파싱 중 파일을 처음 읽으면 Cloud Composer는 DAG를 파싱하는 스케줄러에 파일을 동기화합니다.
DAG 실행: DAG 실행 중에 파일을 처음 읽으면 Cloud Composer가 실행을 실행하는 작업자에게 파일을 동기화합니다.
Airflow 구성요소의 로컬 스토리지가 제한적이므로 구성요소의 디스크 공간을 확보하기 위해 다운로드한 파일을 삭제하는 것이 좋습니다. 동일한 파일을 단일 Airflow 작업자로 다운로드하는 동시 태스크가 있는 경우 로컬 스토리지 사용량이 일시적으로 증가할 수도 있습니다.
로그 폴더
logs/ 폴더는 Cloud Storage API를 사용하여 Airflow 작업자에서 환경 버킷으로 동기화됩니다.
Cloud Storage API 할당량은 이동한 데이터 양으로 계산되므로 시스템에서 실행하는 Airflow 태스크 수에 따라 Cloud Storage API 사용량이 늘어날 수 있습니다. 실행하는 태스크가 많을수록 로그 파일이 커집니다.
웹 서버와 동기화
plugins/ 폴더는 Airflow UI에서 플러그인을 로드할 수 있도록 웹 서버에 자동으로 동기화됩니다.