DAG와 Task의 동시성을 관리하기 위한 변수DAG와 Task의 병렬 처리를 관리하기 위해 사용하는 Airflow 변수는 다음과 같다.parallelism : Scheduler당 동시에 실행할 수 있는 최대 task Instance 수를 정의하며, Worker 수와 무관max_active_tasks_per_dag : 각 DAG에서 동시에 실행하도록 허용된 최대 Task Instance 수max_active_runs_per_dag : DAG당 활성 DAG 실행의 최대 수 이전에 parallelism에 대해서는 글을 작성한 적이 있기에 parallelism에 대해 자세히 알고 싶다면 아래의 링크에서 확인하면 좋을 것 같다. 여기서는 max_active_tasks_per_dag, max_active_runs..
Redshift COPY 시 잘못된 timestamp 값이 적재되는 문제Airflow의 S3ToOperator를 사용해 S3 버킷에 저장된 .parquet 파일을 Redshift에 COPY를 진행하였는데, 다른 모든 값은 모두 올바르게 적재되었음에도 timestamp 형식을 가진 컬럼만 잘못된 값이 적재되었다. 이 문제를 해결하기 위한 과정을 적어보려 한다.데이터 처리 방식 (ETL)우선 이슈가 발생했을 때의 S3에 적재되는 .parquet의 ETL 과정을 나타내 본 뒤 본격적으로 해결 과정을 알아보자.데이터 추출 (Extract)서울 도시데이터 API 호출을 통해 Json 형태의 데이터 추출데이터 변형 (Transform)추출한 데이터 (Json)를 가져와 필요한 데이터를 리스트 형태로 변경여기서 리스..
인프라 구축기현재 인프라를 사용하면서 지속적으로 비용이 청구되고 있다. Redshift Serverless는 가변 비용으로 쿼리 비용만을 청구하며 Bastion Host EC2는 프리티어이므로 비용이 발생하지 않지만, Airflow EC2와 RDS의 경우 서버를 사용하기에 켜두기만 하더라도 비용이 발생한다. 특히 Airflow EC2의 경우 t3.large의 인스턴스를 사용하고 외부와의 통신 (web, API 등)이 자주 이루어지므로 추가로 비용이 발생한다. 그래서 서버를 사용하지 않는 시간에는 서비스를 중지하여 비용을 절약하고자 하였다.사용하지 않을 때 중지할 서비스 : EC2 (Bastion Host, Airflow), RDSRedshfit의 경우 사용한 만큼만 비용이 청구되므로 중지하지 않아도 됨 ..
인프라 구축기AirflowAirflow는 Web server에 Admin 계정으로 접속 시 Security > List Users에서 사용자 관리가 가능하다. 그러나 기본 Admin 계정은 매우 단순하기 때문에 보안을 위해 수정이 필요하다. 그래서 관리 위한 Admin 계정 하나와 3명의 사용자를 위한 User 계정을 생성하려고 한다. 기본 Admin 계정 : ID - airflow, PW - airflow사용자 생성Airflow의 사용자 생성은 UI로 가능하기 때문에 어렵지 않음관리자 계정은 Role = Admin, 사용자(개발자) 계정은 Role = User로 설정역할 별 권한Admin사용자 관리, DAG / Jobs 관리, Variables 및 Connections 관리, 권한 관리, 데이터 관리,..
