버전 제어를 사용하여 DAG 테스트, 동기화, 배포

이 튜토리얼에서는 Cloud Composer 환경의 DAG를 버전 제어 시스템과 동기화하는 CI/CD 흐름을 만드는 방법을 설명합니다. 이 흐름의 순서는 아래와 같습니다.

  1. DAG를 변경하고 해당 변경사항을 저장소의 개발 분기로 푸시합니다.
  2. 저장소의 기본 분기에 대한 pull 요청을 엽니다.
  3. Cloud Build가 단위 테스트를 실행하여 DAG가 유효한지 검사합니다.
  4. pull 요청이 승인되고 기본 분기에 병합됩니다.
  5. Cloud Build가 개발 Cloud Composer 환경을 이러한 새로운 변경사항과 동기화합니다.
  6. DAG가 개발 환경에서 예상대로 작동하는지 확인합니다.
  7. DAG가 예상대로 작동하는 경우 DAG를 프로덕션 Cloud Composer 환경에 수동으로 동기화합니다.
흐름의 단계를 보여주는 아키텍처 다이어그램 사전 제출 및 PR 검토는 GitHub 섹션에 있으며 DAG 동기화 및 수동 DAG 확인은 GCP 섹션에 있습니다.
그림 1. 흐름의 단계를 보여주는 아키텍처 다이어그램 (확대하려면 클릭)

목표

  • Cloud Build를 사용하여 자동화된 DAG 단위 테스트 사전 제출 검사를 실행합니다. 이 검사는 DAG의 단위 테스트를 실행합니다.
  • 개발용 Cloud Composer 환경의 DAG를 버전 제어 시스템의 DAG와 동기화

비용

이 가이드에서는 비용이 청구될 수 있는 다음과 같은 Google Cloud 구성요소를 사용합니다.

프로젝트 사용량을 기준으로 예상 비용을 산출하려면 가격 계산기를 사용하세요. Google Cloud를 처음 사용하는 사용자는 무료 체험판을 사용할 수 있습니다.

이 튜토리얼을 마치면 만든 리소스를 삭제하여 비용이 계속 청구되지 않게 할 수 있습니다. 자세한 내용은 삭제를 참조하세요.

Google Cloud를 처음 사용하는 사용자는 무료 체험판을 사용할 수 있습니다.

시작하기 전에

이 튜토리얼에서는 DAG 및 해당 테스트가 GitHub 저장소에 저장되어 있다고 가정합니다. 예시 저장소의 콘텐츠가 포함된 composer/cicd_sample 디렉터리에서 DAG 및 테스트는 요구사항 파일, 제약조건 파일, 최상위 수준에 저장된 Cloud Build 구성 파일과 함께 dags 폴더에 저장됩니다. DAG 동기화 유틸리티 및 요구사항은 utils 폴더에 있습니다. 이 구조는 Airflow 1, Airflow 2, Cloud Composer 1, Cloud Composer 2 환경에 사용할 수 있습니다.

또한 이 튜토리얼에서는 개발 환경과 프로덕션 환경이라는 두 가지 동일한 Cloud Composer 환경을 사용한다고 가정합니다.

  1. Google Cloud Console의 프로젝트 선택기 페이지에서 Google Cloud 프로젝트를 선택하거나 만듭니다.

    프로젝트 선택기로 이동

  2. Cloud 프로젝트에 결제가 사용 설정되어 있는지 확인합니다. 프로젝트에 결제가 사용 설정되어 있는지 확인하는 방법을 알아보세요.

  3. 개발 및 프로덕션 Cloud Composer 환경을 만들지 않았다면 지금 만드세요. 이 튜토리얼에서는 Cloud Composer 2 환경 또는 Cloud Composer 1 환경을 만들 수 있습니다.

환경 준비

이 섹션에서는 Cloud Build 작업 두 개를 설정합니다. 첫 번째 작업은 DAG의 단위 테스트를 실행하는 사전 제출 검사를 실행합니다. 두 번째 작업은 DAG가 저장소의 기본 분기에 병합된 후 Cloud Composer 환경과 동기화됩니다.

사전 제출 검사 추가

단위 테스트 추가

아직 작성하지 않은 경우 DAG의 단위 테스트를 작성합니다. 이러한 테스트를 각각 _test 서픽스를 붙여 저장소에 DAG와 함께 저장합니다. 예를 들어 example_dag.py의 DAG 테스트 파일은 example_dag_test.py입니다. 이들은 저장소에서 사전 제출 검사로 실행되는 테스트입니다.

Cloud Build 사전 제출 검사 추가

이 섹션에서는 사전 제출 검사를 위한 Cloud Build YAML 구성 및 Cloud Build 트리거를 만듭니다.

Cloud Build YAML 구성 만들기

test-dags.cloudbuild.yaml이라는 Cloud Build 작업을 구성하는 YAML 파일을 만드세요. 이 절차는 3단계로 구성됩니다.

  1. DAG에 필요한 종속 항목 설치
  2. 단위 테스트에 필요한 종속 항목 설치
  3. DAG 테스트 실행

    
    steps:
      # install dependencies
      - name: python:3.8-slim
        entrypoint: pip
        args: ["install", "-r", "requirements.txt", "-c", "constraints.txt", "--user"]
    
      - name: python:3.8-slim
        entrypoint: pip
        args: ["install", "-r", "requirements-test.txt", "--user"]
    
      # run in python 3.8 which is latest version in Cloud Composer
      - name: python:3.8-slim
        entrypoint: python3.8
        args: ["-m", "pytest", "-s", "dags/"]
    
Cloud Build 트리거 만들기

GitHub에서 저장소 빌드에 대한 가이드에 따라 아래의 구성으로 GitHub 앱 기반 트리거를 만듭니다.

  • 이름: test-dags
  • 이벤트: Pull 요청
  • 소스 - 저장소: 저장소 선택
  • 소스 - 기본 분기: ^main$(필요한 경우 main을 저장소의 기본 분기 이름으로 변경)
  • 소스 - 주석 제어: 필수 아님
  • 빌드 구성 - Cloud 빌드 구성 파일: /test-dags.cloudbuild.yaml(빌드 파일의 경로)

DAG 동기화 작업 추가

다음으로 유틸리티 스크립트를 실행하는 Cloud Build 작업을 구성합니다. 이 작업의 유틸리티 스크립트는 저장소의 기본 분기에 병합된 DAG를 Cloud Composer 환경과 동기화합니다.

DAG 유틸리티 스크립트 추가

이 유틸리티 스크립트는 DAG가 아닌 모든 Python 파일을 무시하고 저장소의 dags/ 디렉터리에 있는 모든 DAG 파일을 임시 디렉터리에 복사합니다. 그런 다음 스크립트는 Cloud Storage 클라이언트 라이브러리를 사용하여 해당 임시 디렉터리의 모든 파일을 Cloud Composer 환경의 dags/ 폴더에 업로드합니다.

import argparse
import glob
import os
from shutil import copytree, ignore_patterns
import tempfile
from typing import List, Tuple

# Imports the Google Cloud client library
from google.cloud import storage

def _create_dags_list(dags_directory: str) -> Tuple[str, List[str]]:
    temp_dir = tempfile.mkdtemp()

    # ignore non-DAG Python files
    files_to_ignore = ignore_patterns("__init__.py", "*_test.py")

    # Copy everything but the ignored files to a temp directory
    copytree(dags_directory, f"{temp_dir}/", ignore=files_to_ignore, dirs_exist_ok=True)

    # The only Python files left in our temp directory are DAG files
    # so we can exclude all non Python files
    dags = glob.glob(f"{temp_dir}/*.py")
    return (temp_dir, dags)

def upload_dags_to_composer(dags_directory: str, bucket_name: str) -> None:
    temp_dir, dags = _create_dags_list(dags_directory)

    if len(dags) > 0:
        # Note - the GCS client library does not currently support batch requests on uploads
        # if you have a large number of files, consider using
        # the Python subprocess module to run gsutil -m cp -r on your dags
        # See https://cloud.google.com/storage/docs/gsutil/commands/cp for more info
        storage_client = storage.Client()
        bucket = storage_client.bucket(bucket_name)

        for dag in dags:
            # Remove path to temp dir
            # if/else is a relative directory workaround for our tests
            current_directory = os.listdir()
            if "dags" in current_directory:
                dag = dag.replace(f"{temp_dir}/", "dags/")
            else:
                dag = dag.replace(f"{temp_dir}/", "../dags/")

            # Upload to your bucket
            blob = bucket.blob(dag)
            blob.upload_from_filename(dag)
            print(f"File {dag} uploaded to {bucket_name}/{dag}.")

    else:
        print("No DAGs to upload.")

if __name__ == "__main__":
    parser = argparse.ArgumentParser(
        description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter
    )
    parser.add_argument(
        "--dags_directory",
        help="Relative path to the source directory containing your DAGs",
    )
    parser.add_argument(
        "--dags_bucket",
        help="Name of the DAGs bucket of your Composer environment without the gs:// prefix",
    )

    args = parser.parse_args()

    upload_dags_to_composer(args.dags_directory, args.dags_bucket)

DAG를 동기화하는 Cloud Build 작업 추가

이 섹션에서는 DAG 동기화 작업을 위한 Cloud Build YAML 구성과 Cloud Build 트리거를 만듭니다.

Cloud Build YAML 구성 만들기

add-dags-to-composer.cloudbuild.yaml이라는 Cloud Build 작업을 구성하는 YAML 파일을 만드세요. 이는 다음과 같은 두 단계로 이루어집니다.

  1. DAG 유틸리티 스크립트에 필요한 종속 항목 설치하기
  2. 유틸리티 스크립트를 실행하여 저장소의 DAG를 Cloud Composer 환경과 동기화하기

    steps:
      # install dependencies
      - name: python
        entrypoint: pip
        args: ["install", "-r", "utils/requirements.txt", "--user"]
    
      # run
      - name: python
        entrypoint: python
        args: ["utils/add_dags_to_composer.py", "--dags_directory=${_DAGS_DIRECTORY}", "--dags_bucket=${_DAGS_BUCKET}"]
    
Cloud Build 트리거 만들기

이 가이드에 따라 다음과 같은 구성으로 GitHub 앱 기반 트리거를 만드세요.

  • 이름: add-dags-to-composer
  • 이벤트: 분기로 푸시
  • 소스 - 저장소: 저장소 선택
  • 소스 - 기본 분기: ^main$(필요한 경우 main을 저장소의 기본 분기 이름으로 변경)
  • 소스 - 포함된 파일 필터(glob): dags/**
  • 빌드 구성 - Cloud 빌드 구성 파일: /add-dags-to-composer.cloudbuild.yaml(빌드 파일의 경로)

고급 구성에서 대체 변수 두 개를 추가합니다.

  • _DAGS_DIRECTORY - DAG가 저장소에서 위치하는 디렉터리. 이 튜토리얼을 정확하게 따른 경우 dags/입니다.
  • _DAGS_BUCKET - gs:// 프리픽스 없이 개발 Cloud Composer 환경에 dags/ 폴더가 포함된 Cloud Storage 버킷. 예를 들면 us-east1-my-env-1234ab56-bucket입니다.

종합

이 섹션에서는 새로 만든 Cloud Build 트리거를 활용하는 DAG 개발 흐름을 따릅니다.

사전 제출 작업 실행

기본 분기에 대한 pull 요청을 만들어 빌드를 테스트하세요. 체크 정보는 페이지 하단에 있습니다. Google Cloud Console에서 빌드 로그를 보려면 '세부정보'를 클릭하고 'Google Cloud Build에 대한 추가 세부정보 보기'를 선택하세요.

빨간색 화살표가 괄호 안의 프로젝트 이름을 가리키는 테스트 DAG라는 GitHub 검사 스크린샷
검사 2. GitHub의 Cloud Build 사전 제출 검사 상태 스크린샷(확대하려면 클릭)

사전 제출 검사가 실패하면 빌드 실패 해결 섹션을 참조하세요.

개발 Cloud Composer 환경에서 DAG 성공 검증

pull 요청이 승인되면 기본 분기에 병합합니다. Google Cloud Console을 사용하여 빌드 결과를 확인합니다. Cloud Build 트리거가 많으면 트리거 이름 add-dags-to-composer로 빌드를 필터링할 수 있습니다.

Cloud Build 동기화 작업이 성공하면 DAG가 개발 Cloud Composer 환경에 표시됩니다. 여기에서 DAG가 예상대로 작동하는지 확인할 수 있습니다. DAG가 예상한 대로 실행되면 프로덕션 Cloud Composer 환경의 dags/ 폴더에 수동으로 DAG 파일을 추가하여 프로덕션 환경으로 승격합니다.

DAG 동기화 작업이 실패했거나 개발 Cloud Composer 환경에서 DAG가 제대로 작동하지 않는 경우 빌드 실패 해결 섹션을 참조하세요.

빌드 실패 해결

이 섹션에서는 일반적인 빌드 실패 시나리오를 해결하는 방법을 알아봅니다.

사전 제출 검사에 실패하면 어떻게 되나요?

pull 요청에서 '세부정보'를 클릭하고 'Google Cloud Build에 대한 추가 세부정보 보기'를 선택하여 Google Cloud Console에서 빌드 로그를 확인합니다. 이러한 로그를 사용하여 DAG 문제를 디버깅할 수 있습니다. 문제가 해결되면 수정을 커밋하고 분기로 푸시합니다. 사전 제출 검사가 다시 실행되며 로그를 디버깅 도구로 계속 사용할 수 있습니다.

DAG 동기화 작업이 실패하면 어떻게 되나요?

Google Cloud Console을 사용하여 빌드 결과를 확인합니다. Cloud Build 트리거가 많으면 트리거 이름 add-dags-to-composer로 빌드를 필터링할 수 있습니다. 빌드 작업의 로그를 검토하고 오류를 해결합니다. 오류 해결에 추가 도움이 필요한 경우 지원 채널을 활용하세요.

DAG가 Cloud Composer 환경에서 제대로 작동하지 않는 경우에는 어떻게 하나요?

DAG가 개발 Cloud Composer 환경에서 예상대로 작동하지 않으면 DAG를 프로덕션 Cloud Composer 환경으로 수동으로 승격하지 마세요. 그 대신, 다음 방법 중 하나를 따르세요.

  1. DAG를 중단시킨 변경사항으로 pull 요청을 되돌리기하여 변경 직전의 상태로 복원합니다(이렇게 하면 해당 pull 요청 안의 다른 모든 파일도 되돌리기됩니다).
  2. 새로운 pull 요청을 만들어 손상된 DAG에 대한 변경사항을 수동으로 되돌립니다.
  3. 새로운 pull 요청을 만들어 DAG의 오류를 수정합니다.

이 단계를 수행하면 사전 제출 검사가 다시 트리거되고 병합 시 DAG 동기화 작업이 트리거됩니다.

삭제

이 튜토리얼에서 사용된 리소스 비용이 Google Cloud 계정에 청구되지 않도록 하려면 리소스가 포함된 프로젝트를 삭제하거나 프로젝트를 유지하고 개별 리소스를 삭제하세요.

프로젝트 삭제

  1. Cloud Console에서 리소스 관리 페이지로 이동합니다.

    리소스 관리로 이동

  2. 프로젝트 목록에서 삭제할 프로젝트를 선택하고 삭제를 클릭합니다.
  3. 대화상자에서 프로젝트 ID를 입력한 후 종료를 클릭하여 프로젝트를 삭제합니다.

개별 리소스 삭제

Cloud Build 리소스 삭제

Google Cloud Console 또는 Google Cloud CLI를 사용하여 두 개의 빌드 트리거를 삭제하세요.

  • test-dags
  • add-dags-to-composer

Cloud Composer 리소스 삭제

Cloud Composer 환경 삭제

이 튜토리얼를 위해 Cloud Composer 환경을 만들었지만 계속 사용할 뜻이 없으면 삭제하여 요금이 청구되지 않도록 합니다. 환경을 유지하려면 DAG를 삭제하세요.

Cloud Composer에서 DAG 삭제

개발 및 프로덕션 Cloud Composer 환경에서 DAG를 삭제합니다.

다음 단계