문제 해결: Apache Airflow v2의 DAG, 연산자, 연결 및 기타 문제

AWS Secrets Manager 암호를 사용하여 Apache Airflow 연결 구성에서 Apache Airflow 연결 및 변수를 위한 암호 키를 만드는 방법을 알아봅니다.

Apache Airflow 변수에 AWS Secrets Manager 암호 키 사용에서 Apache Airflow 변수(test-variable)에 대해 암호 키를 사용하는 방법을 알아봅니다.

Apache Airflow 연결을 위한 AWS Secrets Manager의 암호 키 사용에서 Apache Airflow 연결(myconn)에 대해 암호 키를 사용하는 방법을 알아봅니다.

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

사용자 환경의 requirements.txt 파일에 다음 항목을 추가합니다.

apache-airflow-providers-snowflake==1.3.0

DAG에 다음 가져오기를 추가합니다.

from airflow.providers.snowflake.operators.snowflake import SnowflakeOperator

Conn Id: snowflake_conn

Conn Type: Snowflake

Host: <my account>.<my region if not us-west-2>.snowflakecomputing.com

Schema: <my schema>

Login: <my user name>

Password: ********

Port: <port, if any>

추가:

{
    "account": "<my account>",
    "warehouse": "<my warehouse>",
    "database": "<my database>",
    "region": "<my region if not using us-west-2 otherwise omit this line>"
}

Amazon MWAA 환경에 설치된 Apache Airflow 공급자 패키지의 Apache Airflow UI에서 Amazon MWAA가 제공하는 연결 유형을 확인합니다.

Apache Airflow CLI 명령 참조의 CLI에서 Apache Airflow 연결을 생성하는 명령을 확인합니다.

연결 유형 개요의 Amazon MWAA의 Apache Airflow UI에서 사용할 수 없는 연결 유형에 대해 Apache Airflow UI의 연결 템플릿을 서로 바꿔서 사용하는 방법을 알아봅니다.

Apache Airflow 구성 옵션을 확인합니다. Apache Airflow 구성 옵션으로 지정한 키-값 쌍(예: AWS Secrets Manager)이 올바르게 구성되었는지 확인합니다. 자세한 내용은 Secrets Manager에 연결할 수 없음 단원을 참조하십시오.

requirements.txt를 확인합니다. Airflow "extras" 패키지 및 requirements.txt에 나열된 기타 라이브러리가 Apache Airflow 버전과 호환되는지 확인합니다.

requirements.txt 파일에서 Python 종속성을 지정하는 방법을 살펴보려면 requirements.txt에서의 Python 종속성 관리 단원을 참조하십시오.

VPC 보안 그룹이 포트 5432에 대한 인바운드 액세스를 허용하는지 확인합니다. 이 포트는 사용자 환경의 Amazon Aurora PostgreSQL 메타데이터 데이터베이스에 연결하는 데 필요합니다. 이 규칙을 추가한 후 Amazon MWAA에 몇 분 정도 시간을 주면 오류가 사라집니다. 자세한 내용은 Amazon MWAA에서 VPC 보안 단원을 참조하십시오.

스케줄러가 실행되지 않는 경우 종속성 설치 실패 또는 스케줄러 과부하와 같은 여러 요인 때문일 수 있습니다. CloudWatch Logs의 해당 로그 그룹을 확인하여 DAG, 플러그인 및 요구 사항이 제대로 작동하는지 확인합니다. 자세한 내용은 Amazon Managed Workflows for Apache Airflow에 대한 모니터링 및 지표 단원을 참조하십시오.

정의된 DAG 수가 많을 수 있습니다. DAG 수를 줄이고 환경 업데이트(예: 로그 수준 변경)를 수행하여 강제로 재설정합니다.

1. Airflow는 DAG의 활성화 여부를 구문 분석합니다. 환경 용량의 50% 이상을 사용하는 경우 Apache Airflow 스케줄러가 너무 많이 사용되기 시작할 수 있습니다. 이로 인해 CloudWatch 지표의 총 구문 분석 시간이 길어지거나 CloudWatch Logs의 DAG 처리 시간이 길어집니다. Apache Airflow 구성을 최적화하는 다른 방법도 있으나 이 설명서에서는 다루지 않습니다.

2. 환경 성능 조정을 권장하는 모범 사례에 대해 자세히 알아보려면 Amazon MWAA의 Apache Airflow 성능 튜닝 단원을 참조하십시오.

대기열에 작업 수가 많을 수 있습니다. 이는 보통 "None" 상태의 작업 수가 많이 증가하는 것으로 나타나거나 CloudWatch의 대기 중인 작업 및/또는 보류 중인 작업 수가 많을 때 나타납니다. 이는 다음과 같은 이유로 발생할 수 있습니다.

1. 환경에 실행할 수 있는 용량보다 실행할 작업이 많거나 자동 크기 조정 전에 대기열에 대기중인 작업이 많을 경우 작업을 감지하고 추가 작업자를 배포할 시간이 있습니다.

2. 환경에서 실행할 수 있는 용량보다 실행할 작업이 많으면 DAG가 동시에 실행하는 작업 수를 줄이거나 최소 Apache Airflow 작업자를 늘리는 것이 좋습니다.

3. 자동 크기 조정이 추가 작업자를 감지하고 배포할 시간을 갖기 전에 대기열에 대기 중인 작업이 많은 경우에는 작업 배포에 시차를 두거나 최소 Apache Airflow 작업자를 늘리는 것이 좋습니다.

4. AWS Command Line Interface(AWS CLI)의 update-environment 명령을 사용하여 사용자 환경에서 실행되는 최소 또는 최대 작업자 수를 변경할 수 있습니다.

   aws mwaa update-environment --name MyEnvironmentName --min-workers 2 --max-workers 10

5. 환경 성능 조정을 권장하는 모범 사례에 대해 자세히 알아보려면 Amazon MWAA의 Apache Airflow 성능 튜닝 단원을 참조하십시오.

실행 도중에 삭제되는 작업은 Apache Airflow에 더 이상 표시되지 않고 중지되는 작업 로그로 표시될 수 있습니다. 이는 다음과 같은 이유로 발생할 수 있습니다.

1. 1) 현재 작업이 현재 환경 용량을 초과하고, 2) 몇 분 동안 아무 작업도 실행되지 않거나 대기 중인 상태가 지속되면 3) 새 작업이 대기열에 추가되는 경우 이러한 상태가 발생합니다.

2. Amazon MWAA 자동 크기 조정은 작업자를 추가하여 첫 번째 시나리오에 대응합니다. 두 번째 시나리오에서는 추가 작업자를 제거합니다. 대기열에 있는 일부 작업의 경우 작업자가 제거되는 과정으로 이어질 수 있으며 컨테이너가 삭제되면 종료됩니다.

3. 해당 환경의 최소 작업자 수를 늘리는 것이 좋습니다. 또 다른 옵션은 DAG 및 작업의 타이밍을 조정하여 이러한 시나리오가 발생하지 않도록 하는 것입니다.

4. 또한 최소 작업자를 사용자 환경의 최대 작업자와 동일하게 설정하여 자동 크기 조정을 효과적으로 비활성화할 수 있습니다. AWS Command Line Interface(AWS CLI)의 update-environment 명령을 사용하여 최소 및 최대 작업자 수를 동일하게 설정해 자동 크기 조정을 비활성화할 수 있습니다.

   aws mwaa update-environment --name MyEnvironmentName --min-workers 5 --max-workers 5

5. 환경 성능 조정을 권장하는 모범 사례에 대해 자세히 알아보려면 Amazon MWAA의 Apache Airflow 성능 튜닝 단원을 참조하십시오.

작업이 “실행 중” 상태에서 멈춘 경우 작업을 지우거나 성공 또는 실패로 표시할 수도 있습니다. 이렇게 하면 사용자 환경의 자동 확장 구성 요소를 통해 환경에서 실행되는 작업자 수를 줄일 수 있습니다. 다음 이미지는 중단된 작업의 예를 보여줍니다.

1. 문제 있는 작업의 원을 선택한 다음 지우기(그림 참조)를 선택합니다. 이렇게 하면 Amazon MWAA에서 작업자의 규모를 축소할 수 있습니다. 그렇지 않으면 Amazon MWAA에서 어떤 DAG가 활성화되었는지 또는 비활성화되었는지 확인할 수 없으며 대기 중인 작업이 있는 경우 규모를 축소할 수 없습니다.

   

Apache Airflow 참조 가이드의 개념에서 Apache Airflow 작업 수명 주기에 대해 자세히 알아봅니다.

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

"transform" 스크립트를 생성합니다.

#!/bin/bash
cp $1 $2

(선택 사항) macOS 및 Linux 사용자는 스크립트가 실행 가능한지 확인하기 위해 다음 명령을 실행해야 할 수 있습니다.

chmod 777 transform_test.sh

plugins.zip 파일에 스크립트를 추가합니다.

zip plugins.zip transform_test.sh

Amazon S3에 plugins.zip 업로드의 단계를 따릅니다.

Amazon MWAA 콘솔에서 plugins.zip 버전 지정의 단계를 따릅니다.

다음 DAG 파일을 생성합니다.

from airflow import DAG
from airflow.providers.amazon.aws.operators.s3_file_transform import S3FileTransformOperator
from airflow.utils.dates import days_ago
import os

DAG_ID = os.path.basename(__file__).replace(".py", "")

with DAG (dag_id=DAG_ID, schedule_interval=None, catchup=False, start_date=days_ago(1)) as dag:
    file_transform = S3FileTransformOperator(
        task_id='file_transform',
        transform_script='/usr/local/airflow/plugins/transform_test.sh',
        source_s3_key='s3://YOUR_S3_BUCKET/files/input.txt',
        dest_s3_key='s3://YOUR_S3_BUCKET/files/output.txt'
    )

Amazon S3에 DAG 코드 업로드의 단계를 따릅니다.