데이터 엔지니어 기술 블로그

키가 존재하면 업데이트하고, 존재하지 않으면 삽입하는 동작이 필요했다. 하지만 대용량의 요청을 보내면서 무리를 덜 줄 방법을 생각하다가 스택오버플로우에 물어봤는데 금방 답변을 해주었다. 질문했던 내용을 자세히 보려면 이 링크를 참조하면 된다. 기존의 방식 1. 1번 키에 prop_a, prop_b에 데이터 삽입 INSERT INTO example_table (id, prop_a, prop_b) VALUES (1,1,7) ON DUPLICATE KEY UPDATE prop_a=1,prop_b=7; ------------------------------------------------------------- id(primary key), prop_a, prop_b, prop_c 1 1 7 0 --------..

개요 Task는 에어플로우의 기본 실행 단위이다. Task에는 세가지 종류가 있다. 1. Operator: Bash, Python 등과 연결되어 있는 미리 정의된 작업 템플릿 2. Sensor: 외부 이벤트를 기다리는 연산자의 하위 클래스 3. TaskFlow: @task 데코레이터를 사용하여 패키지와 할 수 있는 기능(Airflow 2.0.0 신규 기능) 기본 개념 관계 작업간의 종속성을 설정하기 위해서 upstream, downstream을 사용할 수 있다. 연산자를 사용하여 종속성을 설정할 수 있다. a_task >> b_task >> [c_task, d_task] 작업은 기본적으로 자신의 정보를 전달하지 않으며 독립적으로 실행이 되지만, XCom을 사용하여 다른 태스크로 정보를 전달할 수 있다. ..

개요 에어플로우에서는 execution date 라는 개념이 있다. 과거에 실행하지 못한 작업을 그 시간대에 맞춰서 실행할 때 유용하다. 하지만 사용하면서도 계속 헷갈리는데 이번에 정리해보려고 한다. Execution Date란? 통계청에서 Parameter로 가져오는 날짜를 지정해야 하는 API가 있다고 생각해본다. 이 API를 사용하여 데이터를 수집하는 스크립트를 만들었다. 크론탭으로 이 스크립트를 매일 아침에 실행하게 하였는데 만약 3일 전(2021-06-14) 통계청 서버의 오류로 데이터를 수집하지 못한 것을 알아챘다면 스크립트의 매개변수 중 날짜를 2021-06-14로 설정한 후 다시 실행시켜야 한다. Airflow에는 Execution Date라는 개념이 있어서 매일 데이터를 수집하도록(@da..

개요 Airflow는 복잡한 워크플로우를 프로그래밍 방식으로 작성해서, 스케줄링하고 모니터링할 수 있는 플랫폼이다. 데이터 파이프라인을 이루고 있는 ETL 스크립트들을 스케줄링 할 때 crontab, cloudwatch 등을 사용하는 곳이 많다. 그러나 스크립트들이 많아지고 서로에 대한 의존성이 생기게 되면 컨트롤하기 어렵고, 기존 작업이 실패했을 때 다시 스크립트를 실행하려면 로그를 확인하고 실행해야 하는 등의 문제점이 생긴다. 이러면 바로 복구할수도 없고, 어디서 잘못되었는지 확인하기도 어렵고, 의존성이 있는 스크립트가 잘못되었는데 그 다음 스크립트가 실행이 되버리는 등의 문제점이 발생할 수 있다. Airflow에는 서로에 대한 의존성을 표현할 수 있고, 스크립트가 실패했을 때 알람을 보내 확인하고 ..

개요 에어플로우에서 작업이 실패했을 때 알 수 있는 방법은 Airflow 인터페이스 웹에 들어가서 확인을 하거나 email을 연동하거나 slack을 연동하는 등의 방법이 있다. 현재는 연동이 되어있지 않아서 작업에 실패하면 직접 들어가서 확인을 해야했는데, 이러면 알람이 늦어져서 조치가 어려워진다. 그래서 지금은 현재 사용하는 Slack과 연동을 해보려고 한다. - 이 문서에서 사용하는 에어플로우 버전은 2.0.0이다. Slack으로 메세지 보내는 방법 Airflow Operator 중에 slack으로 메세지를 보내주는 오퍼레이터가 있다. 자세한 내용은 이 링크를 참조하면 된다. 1. Airflow가 설치되어 있는 서버에서 apache-airflow-providers-slack 을 설치한다. python..

Kinesis란? Kinesis는 실시간으로 데이터 스트림을 수집, 처리, 분석해주는 서비스이다. Data Streams 데이터 스트림 수집 및 저장 샤드의 수를 조절하여 스트림을 얼마나 받을지 조절할 수 있다. Data Firehose 데이터 스트림 처리 및 전송 Data Analytics 스트리밍 데이터 분석 실시간 분석 생성 – 지표를 계산하고, Kinesis를 통해 Amazon S3 또는 Amazon Redshift로 전송할 수 있다. 실시간 대시보드 제공 – 집계 및 처리된 스트리밍 데이터 결과를 전송하여 실시간 대시보드를 구성할 수 있다. 실시간 지표 생성 – 실시간 모니터링, 알림, 경보에 사용할 사용자 지정 지표와 트리거를 생성할 수 있다. Video Streams 재생 및 분석을 위해 미..

개요 mysql은 도커 컨테이너로 올라가있는 상태이고, 로컬에 볼륨이 공유된 상태로 사용하고 있었다. mysql에서 로그를 확인해보니 아래와 같은 로그를 확인할 수 있었다. 해결 방법 mbind는 리눅스 명령어로 메모리 범위에 대한 메모리 정책 설정을 뜻한다. mbind를 사용하는 명령어에서 권한이 없는 문제로 예상된다. 도커 공식 문서에 보면 아래와 같은 내용이 있다. --cap-add=sys_nice Grants the container the CAP_SYS_NICE capability, which allows the container to raise process nice values, set real-time scheduling policies, set CPU affinity, and other ..

개요 RDS에서 매일 생성되는 스냅샷으로 Athena에서 과거의 테이블을 사용할 수 있게 했다. 그러나 시간 문제가 두가지가 발생했다. 문제 첫 번째 문제 - 시간으로 필터할 수 없는 문제 S3에 데이터를 저장한 후 그대로 Glue로 크롤링 후 Athena에서 날짜로 필터해서 읽으려고 시도했다. 그러나 쿼리 결과가 비어있었다. Athena는 Java Timestamp 형식이 필요하다고 하는데, 데이터가 그렇지 않은 것으로 예상된다. 해결 방법은 다음과 같다. 열을 스트링으로 정의한다. YYYY-MM-DD HH:MM:SS.fffffffff 형식으로 변환한다. 쿼리에서 Presto의 날짜 및 시간 함수를 사용하여 DATE 또는 TIMESTAMP로 읽는다. https://aws.amazon.com/ko/pre..

개요 Presto는 페이스북에서 아주 많은 데이터를 빠르고 효율적으로 분석하기 위해 개발되었다. 페이스북에서는 300PB 규모의 데이터 웨어하우스에 쿼리할 때 사용하고 있으며 현재는 Airbnb와 Dropbox 등에서도 사용되고 있다. 다양한 소스를 지원해주는데 Hive, Cassandra, RDB, AWS S3 등에서 데이터를 읽을 수 있다. Hive는 처리할 때 중간 결과를 디스크에 저장하지만 Presto는 메모리에 저장하기 때문에 속도가 훨씬 빠르다. 하지만 리소스를 더 사용하게 된다. Architecture Presto는 하나의 Coordinator가 다양한 소스에서 데이터를 가져와 Worker에게 전달을 해주는 역할을 한다. 쿼리 처리 방법 클라이언트에서 Coordinator로 쿼리를 보낸다. ..

개요 This type of correlated subquery pattern is not supported due to internal error 레드시프트에서 쿼리를 보낼 때 위와 같은 에러를 반환할 때가 있다. Redshift는 쿼리 플래너에서 쿼리 재작성을 이용하여 상관관계가 있는 패턴 등을 최적화한다. MySQL과는 처리해주는 방식이 다르기 때문에 지원하지 않는 상관관계를 가진 서브쿼리들이 있다. 지원하지 않는 서브쿼리 1번) 건너뛰기 수준의 상관관계 참조 event 테이블은 가장 상위 레벨에 있으나 가장 하위의 서브쿼리에 event 테이블이 상관관계로 사용되고 있다. select event.eventname from event where not exists ( select * from list..