Cloud Composer 2에서 Apache Airflow DAG 실행

Cloud Composer 2에서 Apache Airflow DAG 실행

Cloud Composer 1 | Cloud Composer 2 | Cloud Composer 3

이 빠른 시작 가이드에서는 Cloud Composer 환경을 만들고 Cloud Composer 2에서 Apache Airflow DAG를 실행하는 방법을 보여줍니다.

• Airflow를 처음 사용하는 경우 Airflow 개념, 객체, 사용법에 대한 자세한 내용은 Apache Airflow 문서의 Airflow 개념 튜토리얼을 참조하세요.

• 대신 Google Cloud CLI를 사용하려면 Cloud Composer에서 Apache Airflow DAG 실행(Google Cloud CLI)을 참조하세요.

• Terraform을 사용하여 환경을 만들려면 환경 만들기(Terraform)를 참조하세요.

시작하기 전에

1. Sign in to your Google Cloud account. If you're new to Google Cloud, create an account to evaluate how our products perform in real-world scenarios. New customers also get $300 in free credits to run, test, and deploy workloads.

2. In the Google Cloud console, on the project selector page, select or create a Google Cloud project.

   Note: If you don't plan to keep the resources that you create in this procedure, create a project instead of selecting an existing project. After you finish these steps, you can delete the project, removing all resources associated with the project.

   Go to project selector

3. Make sure that billing is enabled for your Google Cloud project.

4. In the Google Cloud console, on the project selector page, select or create a Google Cloud project.

   Note: If you don't plan to keep the resources that you create in this procedure, create a project instead of selecting an existing project. After you finish these steps, you can delete the project, removing all resources associated with the project.

   Go to project selector

5. Make sure that billing is enabled for your Google Cloud project.

6. Enable the Cloud Composer API.

   Enable the API

7. 빠른 시작을 완료하는 데 필요한 권한을 얻으려면 관리자에게 프로젝트에 대한 다음 IAM 역할을 부여해 달라고 요청하세요.

   • Cloud Composer 환경의 만들기 및 관리를 확인하려면 다음 안내를 따르세요.
     • 환경 및 스토리지 객체 관리자(roles/composer.environmentAndStorageObjectAdmin)
     • 서비스 계정 사용자(roles/iam.serviceAccountUser)
   • 로그 보기: 로그 뷰어(roles/logging.viewer)

   역할 부여에 대한 자세한 내용은 프로젝트, 폴더, 조직에 대한 액세스 관리를 참조하세요.

   커스텀 역할이나 다른 사전 정의된 역할을 통해 필요한 권한을 얻을 수도 있습니다.

환경의 서비스 계정 만들기

사용자는 환경을 만들 때 서비스 계정을 지정합니다. 이 서비스 계정을 환경의 서비스 계정이라고 합니다. 환경은 이 서비스 계정을 사용하여 대부분의 작업을 실행합니다.

환경의 서비스 계정은 사용자 계정이 아닙니다. 서비스 계정은 사용자가 아닌 애플리케이션 또는 가상 머신 (VM) 인스턴스에서 사용하는 특별한 유형의 계정입니다.

예정된 지원 중단: 서비스 계정을 지정하지 않고도 환경을 만들 수 있지만 권장하지는 않습니다. 이 경우 환경은 기본 Compute Engine 서비스 계정을 사용합니다. 이 계정에는 광범위한 IAM 권한이 있으므로 이 섹션에 설명된 대로 환경에 별도의 서비스 계정을 만드는 것이 좋습니다. 향후 Cloud Composer 출시에서는 새 Cloud Composer 환경이 더 이상 기본적으로 Compute Engine 서비스 계정을 사용하지 않습니다.

환경의 서비스 계정을 만들려면 다음 안내를 따르세요.

1. Identity and Access Management 문서에 설명된 대로 새 서비스 계정을 만듭니다.

2. Identity and Access Management 문서에 설명된 대로 역할을 부여합니다. 필요한 역할은 Composer 작업자 (composer.worker)입니다.

환경 만들기

1. Google Cloud 콘솔에서 환경 만들기 페이지로 이동합니다.

   환경 만들기로 이동

1. 프로젝트의 첫 번째 환경인 경우 Cloud Composer 서비스 계정에 필요한 권한 부여 섹션이 표시됩니다.

   Cloud Composer 서비스 에이전트 계정을 환경 서비스 계정의 새 주 구성원으로 추가하고 Cloud Composer v2 API 서비스 에이전트 확장 역할을 부여합니다.

   원하는 환경 서비스 계정을 사용하고 있는지 확인하고 부여를 클릭합니다.

2. 이름 필드에 example-environment를 입력합니다.

3. 위치 드롭다운 목록에서 Cloud Composer 환경의 리전을 선택합니다. 이 가이드에서는 us-central1 리전을 사용합니다.

4. 다른 환경 구성 옵션에 제공된 기본값을 사용합니다.

5. 만들기를 클릭하고 환경이 생성될 때까지 기다립니다.

   참고: Cloud Composer에서 환경을 만들려면 약 25분 정도 걸립니다.

6. 완료되면 환경 이름 옆에 녹색 체크표시가 나타납니다.

DAG 파일 만들기

Airflow DAG는 예약 및 실행하려는 태스크가 구성된 모음입니다. DAG는 표준 Python 파일에서 정의됩니다.

이 가이드에서는 quickstart.py 파일에 정의된 Airflow DAG 예시를 사용합니다. 이 파일의 Python 코드는 다음을 수행합니다.

1. DAG, composer_sample_dag를 만듭니다. 이 DAG는 매일 실행됩니다.
2. 태스크 하나, print_dag_run_conf를 실행합니다. 태스크는 bash 연산자를 사용하여 DAG 실행 구성을 출력합니다.

quickstart.py 파일의 사본을 로컬 머신에 저장합니다.

import datetime

from airflow import models
from airflow.operators import bash

# If you are running Airflow in more than one time zone
# see https://airflow.apache.org/docs/apache-airflow/stable/timezone.html
# for best practices
YESTERDAY = datetime.datetime.now() - datetime.timedelta(days=1)

default_args = {
    "owner": "Composer Example",
    "depends_on_past": False,
    "email": [""],
    "email_on_failure": False,
    "email_on_retry": False,
    "retries": 1,
    "retry_delay": datetime.timedelta(minutes=5),
    "start_date": YESTERDAY,
}

with models.DAG(
    "composer_quickstart",
    catchup=False,
    default_args=default_args,
    schedule_interval=datetime.timedelta(days=1),
) as dag:
    # Print the dag_run id from the Airflow logs
    print_dag_run_conf = bash.BashOperator(
        task_id="print_dag_run_conf", bash_command="echo {{ dag_run.id }}"
    )

환경의 버킷에 DAG 파일 업로드

모든 Cloud Composer 환경에는 Cloud Storage 버킷이 연결되어 있습니다. Cloud Composer의 Airflow는 이 버킷의 /dags 폴더에 있는 DAG만 예약합니다.

DAG를 예약하려면 로컬 머신에서 환경의 /dags 폴더로 quickstart.py를 업로드합니다.

1. Google Cloud 콘솔에서 환경 페이지로 이동합니다.

   환경으로 이동

2. 환경 목록에서 환경 이름 example-environment를 클릭합니다. 환경 세부정보 페이지가 열립니다.

3. DAG 폴더 열기를 클릭합니다. 버킷 세부정보 페이지가 열립니다.

4. 파일 업로드를 클릭한 후 quickstart.py의 복사본을 선택합니다.

5. 파일을 업로드하려면 열기를 클릭합니다.

DAG 보기

DAG 파일을 업로드하면 Airflow는 다음을 수행합니다.

1. 업로드한 DAG 파일을 파싱합니다. DAG가 Airflow에서 사용 가능해지기까지 몇 분 정도 걸릴 수 있습니다.
2. 사용 가능한 DAG 목록에 DAG를 추가합니다.
3. DAG 파일에 제공한 일정에 따라 DAG를 실행합니다.

DAG UI에서 확인하여 DAG가 오류 없이 처리되고 Airflow에서 사용할 수 있는지 확인합니다. DAG UI는 Google Cloud 콘솔에서 DAG 정보를 볼 수 있는 Cloud Composer 인터페이스입니다. 또한 Cloud Composer는 기본 Airflow 웹 인터페이스인 Airflow UI에 대한 액세스를 제공합니다.

1. Airflow가 이전에 업로드한 DAG 파일을 처리하고 첫 번째 DAG 실행을 완료할 때까지 약 5분 정도 기다립니다(뒷부분에서 설명).

2. Google Cloud 콘솔에서 환경 페이지로 이동합니다.

   환경으로 이동

3. 환경 목록에서 환경 이름 example-environment를 클릭합니다. 환경 세부정보 페이지가 열립니다.

4. DAG 탭으로 이동합니다.

5. composer_quickstart DAG가 DAG 목록에 있는지 확인합니다.

   

   그림 1. DAG 목록에는 composer_quickstart DAG가 표시됩니다(확대하려면 클릭).

   참고: airflow_monitoring DAG는 Cloud Composer에서 환경의 상태를 모니터링하는 데 사용됩니다.

DAG 실행 세부정보 보기

DAG를 한 번 실행하는 것을 DAG 실행이라고 합니다. DAG 파일의 시작 날짜가 어제로 설정되어 있으므로 Airflow는 예시 DAG의 DAG 실행을 즉시 실행합니다. Airflow는 이러한 방식으로 지정된 DAG의 일정을 따라잡습니다.

예시 DAG에는 콘솔에서 echo 명령어를 실행하는 print_dag_run_conf 태스크 하나가 포함되어 있습니다. 이 명령어는 DAG(DAG 실행의 숫자 식별자)에 대한 메타 정보를 출력합니다.

1. DAG 탭에서 composer_quickstart를 클릭합니다. DAG의 실행 탭이 열립니다.

2. DAG 실행 목록에서 첫 번째 항목을 클릭합니다.

   

   그림 2. composer_quickstart DAG의 DAG 실행 목록(확대하려면 클릭)

3. 예시 DAG의 개별 태스크에 대한 정보를 자세히 보여주는 DAG 실행 세부정보가 표시됩니다.

   

   그림 3. DAG 실행에서 실행된 태스크 목록(확대하려면 클릭)

4. DAG 실행 로그 섹션에는 DAG 실행의 모든 태스크 로그가 나열됩니다. 로그에서 echo 명령어의 출력을 볼 수 있습니다.

   

   그림 4. print_dag_run_conf 태스크 로그(확대하려면 클릭)

삭제

이 페이지에서 사용한 리소스 비용이 Google Cloud 계정에 청구되지 않도록 하려면 다음 단계를 수행합니다.

이 튜토리얼에서 사용된 리소스를 삭제합니다.

1. Cloud Composer 환경을 삭제합니다.

   1. Google Cloud 콘솔에서 환경 페이지로 이동합니다.

      환경으로 이동

   2. example-environment를 선택하고 삭제를 클릭합니다.

   3. 환경이 삭제될 때까지 기다립니다.

2. 환경의 버킷을 삭제합니다. Cloud Composer 환경을 삭제하면 버킷이 삭제되지 않습니다.

   1. Google Cloud 콘솔에서 스토리지 > 브라우저 페이지로 이동합니다.

      스토리지 > 브라우저로 이동

   2. 해당 환경의 버킷을 선택하고 삭제를 클릭합니다. 예를 들어 이 버킷의 이름을 us-central1-example-environ-c1616fe8-bucket으로 지정할 수 있습니다.

3. 해당 환경의 Redis 큐의 영구 디스크를 삭제합니다. Cloud Composer 환경을 삭제해도 영구 디스크는 삭제되지 않습니다.

   1. Google Cloud 콘솔에서 Compute Engine > 디스크로 이동합니다.

      디스크로 이동

   2. 환경의 Redis 큐 영구 디스크를 선택하고 삭제를 클릭합니다.

      예를 들어 이 디스크의 이름을 pvc-02bc4842-2312-4347-8519-d87bdcd31115로 지정할 수 있습니다. Cloud Composer 2의 디스크 크기는 항상 Balanced persistent disk 유형이며 크기는 2 GB입니다.

다음 단계

• DAG 추가 및 업데이트
• DAG 작성
• Airflow 구성 재정의
• Airflow UI 액세스