Insights
AI가 자동으로 큐레이션·번역·정리하는 기술 동향 피드입니다.
본 기사는 Apache Airflow를 활용하여 음악 스트리밍 회사의 데이터 웨어하우스 ETL 파이프라인 구축 과정을 다룹니다. S3에서 데이터를 추출하고 Redshift에 적재하며, 스타 스키마로 변환하는 프로덕션급 워크플로우 설계 및 구현 방법을 설명합니다.
AI 자동 생성 콘텐츠
본 콘텐츠는 GitHub AI Tools의 원문을 AI가 자동으로 요약·번역·분석한 것입니다. 원 저작권은 원저작자에게 있으며, 정확한 내용은 반드시 원문을 확인해 주세요.
원문 바로가기프로젝트 설명: 한 음악 스트리밍 회사가 데이터 웨어하우스 ETL(Extract, Transform, Load) 파이프라인에 더 많은 자동화와 모니터링을 도입하고자 합니다. 그들은 이를 달성하는 가장 좋은 도구는 Apache Airflow라는 결론에 이르렀습니다. 저는 이 회사에서 데이터 엔지니어 역할을 맡았고, 데이터 품질 검사를 통합하고 쉬운 백필(backfill)이 가능한 재사용 가능한 프로덕션급 데이터 파이프라인을 구축하는 임무를 받았습니다. 여러 분석가와 데이터 과학자들이 이 파이프라인의 결과물에 의존하며, 이 파이프라인은 소스에서 새 데이터를 가져와 목적지에 결과를 저장하도록 매일 스케줄링되어 실행될 것으로 예상됩니다.
데이터 설명: 소스 데이터는 S3에 존재하며 Amazon Redshift라는 데이터 웨어하우스에서 처리되어야 합니다. 소스 데이터셋은 애플리케이션의 사용자 활동에 대한 JSON 로그와 사용자가 듣는 노래에 대한 JSON 메타데이터로 구성됩니다.
데이터 파이프라인 설계:
전반적으로 이 파이프라인은 다음과 같은 작업을 수행합니다.
dag
Airflow DAG의 구조
설계 목표:
데이터 소비자들의 요구 사항을 바탕으로, 우리의 파이프라인은 다음 지침을 준수해야 합니다:
파이프라인 구현:
Apache Airflow는 DAG(Directed Acyclic Graph) 내에서 워크플로우를 프로그래밍 방식으로 생성하기 위한 Python 프레임워크입니다. 예를 들어, ETL 프로세스 실행, 보고서 생성, 모델 일일 재학습 등이 가능합니다. Airflow UI는 우리가 정의한 DAG를 자동으로 파싱하여 데이터의 이동 및 변환에 대한 자연스러운 표현을 만들어줍니다. DAG란 여러분이 실행하고자 하는 모든 태스크들의 모음이며, 이들이 갖는 관계와 의존성을 반영하는 방식으로 구성됩니다. DAG는 워크플로우를 어떻게 수행할지(how)를 설명하고, Operators는 실제로 무엇을 수행할지(what)를 결정합니다.
기본적으로 Airflow에는 PythonOperator, BashOperator, DummyOperator 등 몇 가지 간단한 내장 Operator가 제공되지만, Airflow는 BaseOperator의 기능을 확장하여 사용자 정의 Operator를 만들 수 있도록 합니다. 이 프로젝트에서는 여러 개의 커스텀 Operator를 개발했습니다.
operators
각 Operator에 대한 설명은 다음과 같습니다:
이러한 각 오퍼레이터의 코드는 plugins/operators 디렉토리에 위치합니다.
파이프라인 스케줄 및 데이터 파티셔닝:
S3에 있는 이벤트 데이터는 연도(2018)와 월(11)로 파티셔닝되어 있습니다. 우리의 임무는 이벤트 json 파일을 증분적으로 로드하고, 이를 전체 파이프라인을 통해 실행하여 노래의 인기도를 계산한 후 그 결과를 다시 S3에 저장하는 것입니다. 이런 방식으로 우리는 파이프라인을 사용하여 매일 최고의 노래들을 자동화된 방식으로 얻을 수 있습니다. 참고로 이것은 사소한 분석이지만, 비슷한 구조를 따르는 다른 복잡한 쿼리들도 상상할 수 있습니다.
S3 입력 이벤트 데이터:
s3://<bucket>/log_data/2018/11/
2018-11-01-events.json
2018-11-02-events.json
...
S3 출력 노래 인기도 데이터:
s3://skuchkula-topsongs/
songpopularity_2018-11-01
songpopularity_2018-11-02
...
DAG는 start_date, end_date 및 위에서 언급한 다른 설계 선택 사항을 지정하는 일부 default_args를 제공함으로써 구성할 수 있습니다.
default_args = {
'owner': 'shravan',
'start_date': datetime(2018, 11, 1),
...
단계 1: 콘솔 또는 create-redshift-cluster의 노트북을 통해 AWS Redshift 클러스터 생성
AWS Redshift 클러스터를 생성하려면 노트북을 실행하세요. 다음 사항에 유의하세요:
단계 2: Apache Airflow 시작
docker-compose.yml이 포함된 디렉토리에서 docker-compose up을 실행하세요. DAG가 있는 위치를 가리키도록 볼륨 매핑을 했는지 확인하십시오.
참고: Mac에서 Apache Airflow를 관리하는 방법에 대한 자세한 내용은 여기에서 찾을 수 있습니다: https://gist.github.com/shravan-kuchkula/a3f357ff34cf5e3b862f3132fb599cf3
start_airflow
단계 3: Apache Airflow Hook 구성
왼쪽에는 S3 connection이 있습니다. 로그인(Login)과 비밀번호(password)는 생성한 IAM 사용자의 액세스 키와 시크릿 키입니다. 기본적으로 이 자격 증명(credentials)을 사용하여 S3에서 데이터를 읽어올 수 있습니다.
오른쪽은 redshift connection입니다. 이 값들은 Redshift 클러스터에서 쉽게 얻을 수 있습니다.
connections
4단계: create-tables-dag 실행
이 DAG는 스테이징(staging), 팩트(fact) 및 디멘션(dimension) 테이블을 생성할 것입니다. 이 작업을 수동으로 트리거해야 하는 이유는 메인 DAG에서 분리하고 싶기 때문입니다. 일반적으로 테이블 생성은 스크립트를 트리거하는 것만으로 처리될 수 있습니다. 하지만 설명의 목적으로 제가 이를 위한 DAG를 만들고 Airflow가 그 DAG를 트리거하도록 했습니다. 이 DAG는 완료되면 비활성화할 수 있습니다. 이 DAG를 실행한 후에는 AWS Redshift에 모든 테이블이 생성된 것을 확인할 수 있어야 합니다.
5단계: load_and_transform_data_in_redshift dag 활성화
실행 시작 날짜(execution start date)가 2018-11-01이고 스케줄 간격(schedule interval)이 @daily이며 실행 종료 날짜(execution end date)가 2018-11-30이기 때문에, Airflow는 30회 동안 매일 자동으로 DAG를 트리거하고 예약할 것입니다. 아래에는 start_date부터 end_date까지, Airflow에 의해 하루에 한 번씩 트리거된 30개의 DAG 실행(DAG runs)이 표시되어 있습니다.
schedule