DAG 추가 또는 업데이트 - Amazon Managed Workflows for Apache Airflow
사전 조건작동 방법v2에서 변경된 사항Amazon MWAA CLI 유틸리티를 사용한 DAG 테스트Amazon S3에 DAG 코드 업로드DAG 폴더의 경로 지정Apache Airflow UI에서 변경 사항 보기다음 단계

기계 번역으로 제공되는 번역입니다. 제공된 번역과 원본 영어의 내용이 상충하는 경우에는 영어 버전이 우선합니다.

DAG 추가 또는 업데이트

방향성 비순환 그래프(DAG)는 DAG 구조를 코드로 정의하는 Python 파일 내에 정의됩니다. AWS CLI또는 Amazon S3 콘솔을 사용하여 DAGs를 환경에 업로드할 수 있습니다. 이 주제에서는 Amazon S3 버킷의 dags 폴더를 사용하여 Amazon Managed Workflows for Apache Airflow 환경에서 Apache Airflow DAGs를 추가하거나 업데이트하는 단계를 설명합니다.

사전 조건

이 페이지의 단계를 완료하려면 먼저 다음이 필요합니다.

  • 권한 - 관리자가 환경에 대한 AmazonMWAAFullConsoleAccess 액세스 제어 정책에 대한 액세스 권한을 AWS 계정에 부여해야 합니다. 또한 Amazon MWAA 환경은 실행 역할이 환경에서 사용하는 AWS 리소스에 액세스할 수 있도록 허용해야 합니다.

  • 액세스 — 종속성을 웹 서버에 직접 설치하기 위해 퍼블릭 리포지토리에 액세스해야 하는 경우 퍼블릭 네트워크 웹 서버 액세스로 환경을 구성해야 합니다. 자세한 내용은 Apache Airflow 액세스 모드 단원을 참조하십시오.

  • Amazon S3 구성 — DAG, plugins.zip의 사용자 지정 플러그인 및 requirements.txt의 Python 종속성을 저장하는 데 사용되는 Amazon S3 버킷은 퍼블릭 액세스가 차단되고 버전 관리가 활성화된 상태로 구성되어야 합니다.

작동 방법

방향비순환 그래프(DAG)는 DAG의 구조를 코드로 정의하는 단일 Python 파일 내에 정의됩니다. 다음 항목으로 구성됩니다.

Amazon MWAA 환경에서 Apache Airflow 플랫폼을 실행하려면 DAG 정의를 스토리지 버킷의 dags 폴더에 복사해야 합니다. 예를 들어 스토리지 버킷의 DAG 폴더는 다음과 같을 수 있습니다.

예 DAG 폴더
dags/
└ dag_def.py

Amazon MWAA는 Amazon S3 버킷의 새 객체 및 변경된 객체를 Amazon MWAA 스케줄러 및 작업자 컨테이너의 /usr/local/airflow/dags 폴더로 30초마다 자동으로 동기화하여 파일 유형에 상관없이 Amazon S3 소스의 파일 계층 구조를 유지합니다. 새 DAG가 Apache Airflow UI에 나타나는 데 걸리는 시간은 scheduler.dag_dir_list_interval에 의해 제어됩니다. 기존 DAG에 대한 변경 사항은 다음 DAG 처리 루프에서 적용됩니다.

참고

DAG 폴더에 airflow.cfg 구성 파일을 포함할 필요는 없습니다. Amazon MWAA 콘솔에서 기본 Apache Airflow 구성을 재정의할 수 있습니다. 자세한 내용은 Amazon MWAA에서 Apache Airflow 구성 옵션 사용 단원을 참조하십시오.

v2에서 변경된 사항

  • 신규: 연산자, 후크, 실행자. DAG의 가져오기 문과 Amazon MWAA에서 사용자가 plugins.zip에 지정한 사용자 지정 플러그인이 Apache Airflow v1과 Apache Airflow v2 사이에서 변경되었습니다. 예를 들어, Apache Airflow v1의 from airflow.contrib.hooks.aws_hook import AwsHook이 Apache Airflow v2의 from airflow.providers.amazon.aws.hooks.base_aws import AwsBaseHook로 변경되었습니다. 자세한 내용은 Apache Airflow 참조 가이드의 Python API 참조를 참조하십시오.

Amazon MWAA CLI 유틸리티를 사용한 DAG 테스트

  • 명령줄 인터페이스(CLI) 유틸리티는 Amazon Managed Workflows for Apache Airflow 환경을 로컬로 복제합니다.

  • CLI는 Amazon MWAA 프로덕션 이미지와 유사한 Docker 컨테이너 이미지를 로컬로 구축합니다. 이를 통해 Amazon MWAA에 배포하기 전에 로컬 Apache Airflow 환경을 실행하여 DAG, 사용자 지정 플러그인 및 종속성을 개발하고 테스트할 수 있습니다.

  • CLI를 실행하려면 GitHub의 aws-mwaa-local-runner를 참조하십시오.

Amazon S3에 DAG 코드 업로드

Amazon S3 콘솔 또는 AWS Command Line Interface (AWS CLI)를 사용하여 DAG 코드를 Amazon S3 버킷에 업로드할 수 있습니다. 다음 단계에서는 Amazon S3 버킷의 이름이 지정된 dags 폴더에 코드(.py)를 업로드한다고 가정합니다.

사용 AWS CLI

AWS Command Line Interface (AWS CLI)는 명령줄 셸의 명령을 사용하여 AWS 서비스와 상호 작용할 수 있는 오픈 소스 도구입니다. 이 페이지에서 단계를 완료하려면 다음이 필요합니다.

를 사용하여 업로드하려면 AWS CLI

  1. 다음 명령을 사용하여 Amazon S3 버킷을 모두 나열합니다.

    aws s3 ls

  2. 다음 명령을 사용하여 사용자 환경의 Amazon S3 버킷에 있는 파일과 폴더를 나열합니다.

    aws s3 ls s3://YOUR_S3_BUCKET_NAME

  3. 다음 명령은 dag_def.py 파일을 dags 폴더에 업로드합니다.

    aws s3 cp dag_def.py s3://YOUR_S3_BUCKET_NAME/dags/

    이름이 지정된 dags 폴더가 Amazon S3 버킷에 아직 없는 경우 이 명령은 dags 폴더를 생성하고 이름이 지정된 dag_def.py 파일을 새 폴더에 업로드합니다.

Amazon S3 콘솔 사용

Amazon S3 콘솔은 Amazon S3 버킷의 리소스를 생성 및 관리할 수 있는 웹 기반 사용자 인터페이스입니다. 다음 단계에서는 이름이 지정된 dags DAG 폴더가 있다고 가정합니다.

Amazon S3 콘솔을 사용하여 업로드하려면

  1. Amazon MWAA 콘솔에서 환경 페이지를 엽니다.

  2. 환경을 선택합니다.

  3. S3 창의 DAG 코드에서 S3 버킷 링크를 선택하여 Amazon S3 콘솔에서 스토리지 버킷을 엽니다.

  4. dags 폴더를 선택합니다.

  5. 업로드를 선택합니다.

  6. 파일 추가를 선택합니다.

  7. dag_def.py의 로컬 사본을 선택하고 업로드를 선택합니다.

Amazon MWAA 콘솔에서 DAG 폴더의 경로 지정(처음)

다음 단계에서는 이름이 지정된 dags Amazon S3 버킷의 폴더에 대한 경로를 지정한다고 가정합니다.

  1. Amazon MWAA 콘솔에서 환경 페이지를 엽니다.

  2. DAG를 실행할 환경을 선택합니다.

  3. 편집을 선택합니다.

  4. Amazon S3의 DAG 코드 창에서 DAG 폴더 필드 옆에 있는 S3 찾아보기를 선택합니다.

  5. dags 폴더를 선택합니다.

  6. 선택을 선택합니다.

  7. 다음, 환경 업데이트를 선택합니다.

Apache Airflow UI에서 변경 사항 보기

Apache Airflow에 로그인

Apache Airflow UI를 보려면 AWS Identity and Access Management (IAM)의 AWS 계정에 대한 Apache Airflow UI 액세스 정책: AmazonMWAAWebServerAccess 권한이 필요합니다.

Apache Airflow UI에 액세스하려면

  1. Amazon MWAA 콘솔에서 환경 페이지를 엽니다.

  2. 환경을 선택합니다.

  3. Airflow UI 열기를 선택합니다.

다음 단계

  • GitHub의 aws-mwaa-local-runner를 사용하여 DAG, 사용자 지정 플러그인, Python 종속성을 로컬에서 테스트합니다.