Airflow DAG 테스트

Cloud Composer 1 | Cloud Composer 2

DAG를 프로덕션에 배포하기 전에 Airflow CLI 하위 명령어를 실행하여 DAG가 실행되는 것과 동일한 컨텍스트에서 DAG 코드를 파싱할 수 있습니다.

Composer 로컬 개발 CLI 도구를 사용하여 로컬에서 DAG 테스트

Composer 로컬 개발 CLI 도구는 Airflow 환경을 로컬에서 실행하여 Cloud Composer 2의 Apache Airflow DAG 개발을 간소화합니다. 이 로컬 Airflow 환경은 특정 Cloud Composer 2 버전의 이미지를 사용합니다.

이 로컬 Airflow 환경을 사용하여 DAG를 개발하고 테스트한 다음 DAG를 테스트 Cloud Composer 환경으로 전송할 수 있습니다. 이 가이드의 나머지 부분에서는 Cloud Composer 환경에서 테스트 DAG를 설명합니다.

DAG 생성 중 테스트

로컬로 단일 작업 인스턴스를 실행하고 로그 출력을 볼 수 있습니다. 출력을 보면 구문 및 작업 오류를 확인할 수 있습니다. 로컬로 테스트하면 종속 항목을 확인하거나 데이터베이스에 상태를 전달하지 않습니다.

DAG는 테스트 환경의 data/test 폴더에 두는 것이 좋습니다.

테스트 디렉터리 만들기

환경 버킷에서 테스트 디렉터리를 만들고 여기에 DAG를 복사합니다.

gsutil cp -r BUCKET_NAME/dags \
  BUCKET_NAME/data/test

다음을 바꿉니다.

  • BUCKET_NAME: Cloud Composer 환경과 연결된 버킷의 이름입니다.

예:

gsutil cp -r gs://us-central1-example-environment-a12bc345-bucket/dags \
  gs://us-central1-example-environment-a12bc345-bucket/data/test

DAG 업로드에 대한 자세한 내용은 DAG 추가 및 업데이트를 참조하세요.

구문 오류가 있는지 확인

/data/test 폴더에 업로드한 DAG에서 구문 오류를 확인하려면 다음 gcloud 명령어를 입력합니다.

gcloud composer environments run \
  ENVIRONMENT_NAME \
  --location ENVIRONMENT_LOCATION \
  dags list -- --subdir /home/airflow/gcs/data/test

다음을 바꿉니다.

  • ENVIRONMENT_NAME: 환경의 이름입니다.
  • ENVIRONMENT_LOCATION: 환경이 위치한 리전입니다.

태스크 오류가 있는지 확인

/data/test 폴더에 업로드한 DAG에서 태스크별 오류를 확인하려면 다음 gcloud 명령어를 실행합니다.

gcloud composer environments run \
  ENVIRONMENT_NAME \
  --location ENVIRONMENT_LOCATION \
  tasks test -- --subdir /home/airflow/gcs/data/test \
  DAG_ID TASK_ID \
  DAG_EXECUTION_DATE

다음을 바꿉니다.

  • ENVIRONMENT_NAME: 환경의 이름입니다.
  • ENVIRONMENT_LOCATION: 환경이 위치한 리전입니다.
  • DAG_ID: DAG의 ID입니다.
  • TASK_ID: 태스크 ID입니다.
  • DAG_EXECUTION_DATE: DAG의 실행 날짜입니다. 이 날짜는 템플릿 용도로 사용됩니다. 여기에서 지정한 날짜와 관계없이 DAG는 즉시 실행됩니다.

예:

gcloud composer environments run \
  example-environment \
  --location us-central1 \
  tasks test -- --subdir /home/airflow/gcs/data/test \
  hello_world print_date 2021-04-22

배포된 DAG 업데이트 및 테스트

테스트 환경에서 DAG의 업데이트를 테스트하려면 다음 안내를 따르세요.

  1. 업데이트하려는 배포된 DAG를 data/test로 복사합니다.
  2. DAG를 업데이트합니다.
  3. DAG를 테스트합니다.
    1. 구문 오류가 있는지 확인합니다.
    2. 작업 관련 오류가 있는지 확인합니다.
  4. DAG가 정상적으로 실행되는지 확인합니다.
  5. 테스트 환경에서 DAG를 사용 중지합니다.
    1. Airflow UI > DAG 페이지로 이동합니다.
    2. 수정 중인 DAG가 계속 실행되는 경우 DAG를 사용 중지합니다.
    3. 진행 중인 작업을 빠르게 처리하려면 작업 및 성공 표시를 클릭하세요.
  6. DAG를 프로덕션 환경에 배포합니다.
    1. 프로덕션 환경에서 DAG를 사용 중지합니다.
    2. 프로덕션 환경의 dags/ 폴더에 업데이트된 DAG를 업로드합니다.

DAG 테스트 FAQ

프로덕션 환경과 테스트 환경에서 DAG 실행을 어떻게 분리하나요?

예를 들어 Airflow에는 모든 DAG 실행이 공유하는 dags/ 폴더에 소스 코드의 전역 저장소가 있습니다. 실행 중인 DAG를 방해하지 않고 프로덕션 또는 테스트에서 소스 코드를 업데이트하려 합니다.

Airflow는 강력한 DAG 분리를 제공하지 않습니다. 테스트 DAG가 프로덕션 DAG를 방해하지 않도록 별도의 프로덕션을 유지하고 Cloud Composer 환경을 테스트하는 것이 좋습니다.

서로 다른 GitHub 브랜치에서 통합 테스트를 실행할 때 DAG 간섭을 방지하려면 어떻게 해야 하나요?

간섭을 방지하려면 고유한 태스크 이름을 사용하세요. 예를 들어 태스크 ID 앞에 브랜치 이름을 붙일 수 있습니다.

Airflow와의 통합 테스트에 대한 권장사항은 무엇인가요?

Airflow와의 통합 테스트에 전용 환경을 사용하는 것이 좋습니다. DAG 실행 성공을 위한 한 가지 방법은 Cloud Storage 폴더의 파일에 데이터를 쓰고 자체 통합 테스트 사례에서 콘텐츠를 확인하는 것입니다.

다른 DAG 참여자와 효율적으로 공동작업하려면 어떻게 해야 하나요?

각 참여자는 개발에 필요한 data/ 폴더에 하위 디렉터리를 가질 수 있습니다.

data/ 폴더에 추가된 DAG는 Airflow 스케줄러 또는 웹 서버에서 자동으로 선택되지 않습니다.

DAG 참여자는 gcloud composer environments run 명령어와 test 하위 명령어를 --subdir 플래그와 함께 사용하여 수동 DAG 실행을 만들어 참여자의 개발 디렉터리를 지정할 수 있습니다.

예를 들면 다음과 같습니다.

gcloud composer environments run test-environment-name \
  tasks test -- dag-id task-id execution-date \
  --subdir /home/airflow/gcs/data/alice_dev

배포 환경과 프로덕션 환경을 어떻게 동기화하나요?

액세스를 관리하려면 다음 안내를 따르세요.

  • 인증에는 서비스 계정을 사용합니다.

  • 액세스 제어에는 Identity and Access Management 및 Cloud Composer 역할과 권한을 사용합니다.

개발 환경에서 프로덕션 환경으로 배포하려면 다음 안내를 따르세요.

  • 환경 변수 및 PyPI 패키지와 같은 구성을 일관되게 유지합니다.

  • DAG 인수를 일관되게 유지합니다. 하드 코딩을 방지하려면 Airflow 매크로 및 변수를 사용하는 것이 좋습니다.

    예를 들면 다음과 같습니다.

    gcloud composer environments run test-environment-name \
  variables set -- DATA_ENDPOINT_KEY DATA_ENDPOINT_VALUE

다음 단계