[AWS Service] MWAA와 Glue를 연동하여 데이터 파이프라인 구축하기

안녕하세요! 오늘은 Amazon MWAA (Managed Workflows for Apache Airflow)와 AWS Glue를 연동하여 데이터 파이프라인을 구축하는 방법에 대해 알아보겠습니다. MWAA는 Airflow를 관리형 서비스로 제공하여 데이터 엔지니어링 워크플로우를 쉽게 관리할 수 있도록 해줍니다. Glue는 서버리스 ETL(추출, 변환, 로드) 서비스로, 데이터를 정리하고 변환하는 데 사용됩니다.

1. S3 버킷 생성

먼저, S3 버킷을 두 개 생성합니다.

• mwaa-scripts-s3: MWAA 스크립트, DAG, 플러그인, 요구 사항 파일을 저장하는 데 사용됩니다.
• mwaa-dataset-s3: 처리할 CSV 파일을 업로드할 데이터셋 저장소로 사용됩니다.

2. 로컬 환경 설정 (MWAA Local Runner)

Amazon MWAA에서 바로 작업을 시작하기 전에 로컬 환경에서 코드를 테스트하는 것이 좋습니다. 다음 단계를 따라 로컬 환경을 설정합니다.

2.1. Bastion 호스트 설정

원하는 인스턴스 타입을 사용하여 인스턴스를 사용합니다. 다음 명령을 실행하여 Docker와 Git을 설치합니다.

sudo dnf install -y docker
sudo systemctl enable --now docker
sudo usermod -a -G docker ec2-user
sudo su - ec2-user
sudo dnf install -y git

2.2. MWAA Local Runner 클론 및 설정

GitHub에서 aws-mwaa-local-runner 리포지토리를 클론하고 디렉터리로 이동합니다.

git clone https://github.com/aws/aws-mwaa-local-runner.git
cd aws-mwaa-local-runner

Docker 이미지를 빌드합니다.

./mwaa-local-env build-image

로컬 서버를 시작합니다.

./mwaa-local-env start

3. Airflow DAG 작성

Airflow DAG 파일을 작성하여 S3 파일 감지 후 Glue 작업을 트리거하도록 설정합니다. 아래는 예시 DAG 코드입니다 (/dags/etl-dags.py).

from airflow import DAG
from airflow.providers.amazon.aws.operators.glue import GlueJobOperator
from airflow.providers.amazon.aws.sensors.s3 import S3KeySensor
from datetime import datetime, timedelta

default_args = {
    "owner": "airflow",
    "depends_on_past": False,
    "start_date": datetime(2024, 2, 15),
    "email_on_failure": False,
    "email_on_retry": False,
    "retries": 1,
    "retry_delay": timedelta(minutes=5),
}

dag = DAG(
    "s3_trigger_glue_jobs",
    default_args = default_args,
    schedule_interval = None,
    catchup = False
)

s3_sensor = S3KeySensor(
    task_id = "s3_file_sensor",
    bucket_name = "mwaa-dataset-s3-0410",
    bucket_key = "raw/*.csv",
    wildcard_match = True,
    aws_conn_id="aws_default",
    timeout=600,
    poke_interval=30,
    dag=dag,
)

glue_task = GlueJobOperator(
    task_id = "glue_job",
    job_name = "mwaa-job",
    script_location=f"s3://mwaa-scripts-s3-0410/scripts/glue.py",
    s3_bucket="mwaa-scripts-s3-0410",
    create_job_kwargs={"GlueVersion": "4.0", "NumberOfWorkers": 2, "WorkerType": "G.1X"}

)

s3_sensor >> glue_task

4. Glue Job 생성

mwaa-job이라는 이름을 가진 job을 생성합니다.

AWS Glue에서 생성한 스크립트를 Bastion 호스트의 /scripts/glue.py에 저장합니다. 이 스크립트는 MWAA에서 트리거될 Glue 작업을 정의합니다.

5. S3 버킷에 파일 업로드

다음 명령을 사용하여 스크립트, DAG, 요구 사항 파일 등을 mwaa-scripts-s3 버킷에 업로드합니다.

aws s3 cp scripts/ s3://mwaa-scripts-s3-0410/scripts --recursive
aws s3 cp dags/ s3://mwaa-scripts-s3-0410/dags/ --recursive
aws s3 cp requirements/ s3://mwaa-scripts-s3-0410/requirements/ --recursive
aws s3 cp plugins/ s3://mwaa-scripts-s3-0410/plugins/ --recursive

6. MWAA 환경 생성 및 설정

AWS 콘솔에서 MWAA 환경을 생성합니다.

• MWAA가 사용할 IAM 역할에 Admin 권한을 부여해야 합니다.

7. Airflow UI에서 DAG 실행

Airflow UI에 접속하여 DAG를 트리거합니다. "Trigger DAG" 버튼을 클릭하여 DAG를 실행합니다.

8. 성공 확인

Airflow UI에서 DAG 실행 상태를 확인합니다. 모든 작업이 성공적으로 완료되면 다음과 유사한 화면이 표시됩니다.

결론

이 가이드에서는 Amazon MWAA와 AWS Glue를 연동하여 데이터 파이프라인을 구축하는 방법을 설명했습니다. 이 구성을 통해 데이터 워크플로우를 자동화하고 확장 가능한 데이터 처리 시스템을 구축할 수 있습니다.