Kwin's Development Diary.

외부 데이터를 수집하는 파이프라인을 개발하면서 한 가지 고민이 생겼다. API 요청도 정상적으로 끝났고, 파싱 과정에서도 예외가 발생하지 않았다.데이터베이스 적재까지 완료됐기 때문에 Airflow에서도 Task는 success 상태였다. 그런데 실제 적재된 데이터를 확인해보면 상황이 달라질 수 있다. 평소 1,000건 정도 수집되던 데이터가 갑자기 10건밖에 수집되지 않았거나, 필수값이 비어 있거나, 정상적으로 존재해야 하는 참조 코드가 잘못되어 있을 수도 있다. HTTP 요청: 성공파싱: 성공DB 적재: 성공Airflow Task: success하지만 수집 데이터는 비정상코드의 실행만 놓고 보면 파이프라인은 성공했다. 그렇다면 데이터 파이프라인에서 '성공'은 어디까지를 의미하는 걸까? 이번에는 코드가 ..

Airflow를 사용하다 보면 TaskInstance의 상태를 자주 확인하게 된다. 정상적으로 실행되고 있는지 확인하기 위해 running 상태를 보고, 실행이 지연되고 있다면 scheduled나 queued 상태에 머물러 있는지 확인하고 실행에 실패하면 failed나 up_for_retry 상태를 확인하면서 문제의 원인을 찾는다. 나 역시 Airflow를 운영하면서 이런 상태들을 계속 확인해왔지만,막상 각각의 상태가 Airflow 내부에서 어떤 단계에 해당하는지 설명하려고 하니 정확하게 알고 있지 못했다. 그래서 이번에는 실제로 Task가 실행되기까지 어떤 일이 일어나는지 알아보려고 한다.@task로 작성한 함수는 언제 실행될까?TaskFlow API를 사용하면 일반 Python 함수에 @task를 붙..

이전에는 외부 API를 호출하는 Extract Task에서 Retry를 어떻게 설정해야 하는지 정리해봤다. 처음에는 실패한 Task를 다시 실행하는 것만 생각했지만, 여러 Task가 같은 외부 API를 사용하고 있다면 Retry가 오히려 외부 서버의 부하를 증가시킬 수도 있었다. 그래서 Retry 횟수뿐만 아니라 Backoff를 통해 언제 다시 요청할 것인지도 생각해야 했고, 글을 정리하면서 마지막에는 또 다른 의문이 남았다. 여러 DAG가 같은 외부 API를 호출하고 있다면 전체 요청량은 어떻게 제한해야 할까? Airflow에는 Pool이라는 기능이 있다 나도 외부 API를 사용하는 Task의 동시 실행 수를 제한할 때 Pool을 사용해왔기 때문에 대략적인 사용 방법은 알고 있었다. 하지만 막상 다시 ..

이전에는 Airflow Task에서 데이터를 Load하던 중 실패했을 때 Retry가 발생하면 어떤 문제가 생길 수 있는지 정리해봤다.[Apache Airflow] Task Retry, 다시 실행해도 안전한가? - Load [Apache Airflow] Task Retry, 다시 실행해도 안전한가? - LoadAirflow로 데이터 수집 파이프라인을 만들다 보면 외부 API 장애나 네트워크 문제처럼 일시적인 오류가 발생할 수 있다. 이런 경우를 대비해서 대부분의 Task에 Retry를 설정해서 사용해왔다.default_argkwin0825.tistory.com DB에 일부 데이터가 이미 저장된 상태에서 Task가 다시 실행될 수 있기 때문에 단순히 Retry를 설정하는 것만으로는 안전한 재처리를 보장할 ..

Airflow로 데이터 수집 파이프라인을 만들다 보면 외부 API 장애나 네트워크 문제처럼 일시적인 오류가 발생할 수 있다. 이런 경우를 대비해서 대부분의 Task에 Retry를 설정해서 사용해왔다.default_args = { "retries": 3, "retry_delay": timedelta(minutes=5),}실패하면 일정 시간이 지난 뒤 다시 실행되기 때문에 처음에는 Retry를 설정해두면 장애가 발생하더라도 어느 정도 안전하게 처리할 수 있다고 생각했다.(여기서 말하는 Retry는 DAG 전체를 다시 실행하는 것이 아니라, 실패한 Task를 다시 실행하는 Task Retry를 의미한다) 그렇다면 실패한 Task가 다시 실행될 때, 첫 번째 실행에서 이미 처리된 데이터는 어떻게 되는..

데이터 엔지니어로 일하면서 Python, Airflow, PostgreSQL, Kafka 등 여러 기술을 사용해왔다.업무에서 계속 사용하는 기술이다 보니 자연스럽게 어느 정도는 알고 있다고 생각했다. 직접 데이터 파이프라인을 만들고 운영하면서 문제가 발생하면 원인을 찾아 해결하기도 했으니, 적어도 내가 사용하고 있는 기술에 대해서는 어느 정도 이해하고 있다고 생각했던 것 같다. 그런데 최근 문득 내가 데이터 엔지니어로서 어느 정도 수준인지 궁금해졌다.경력이 오래된 것도 아니고, 그렇다고 이제 막 일을 시작한 것도 아니다 보니 스스로 객관적으로 판단하기가 쉽지 않았다.그래서 새로운 기술을 더 공부하기 전에 지금까지 사용해온 기술들을 내가 어느 정도까지 이해하고 있는지 하나씩 점검해보기로 했다.생각보다 익숙..

collections.defaultdict: 파이썬 collections 모듈에 있는 딕셔너리 서브 클래스이다.: 일반 dict와 거의 동일하게 동작하지만, 존재하지 않는 키에 접근했을 때 자동으로 기본값을 생성하여 넣어주는 기능이 있다.from collections impor defaultdict - dict: 존재하지 않는 키에 접근시 KeyError 발생- defaultdict: 존재하지 않는 키에 접근시 미리 지정한 함수를 호출하여 기본값을 만든 뒤, 그 값을 해당 키에 저장하고 반환 사용법생성 방법from collections import defaultdict# 기본값으로 설정할 자료형을 객체 생성시 선언해준다# int: 0d = defaultdict(int)# list: []d = defaul..

[ 지원서 제출 ]2024년 3월, 당시 학부생 7번째 학기(4학년 1학기)가 막 시작되어 열심히 살겠다는 생각은 언제나 가슴 한편에 자리하고 있었지만 열심히 사는 것이 대체 어떤 것인지 몰라 답답하던 참이었다. 그러던 중 학교 벽면에 붙은 카카오테크 캠퍼스 2기(이후 카테캠)를 모집한다는 공고를 우연히 보게 되었지만, 카카오라는 네임벨류에 겁먹어 내가 감히 지원해도 될까라는 생각에 지원할 생각을 하지 못했다. 이후 카테캠 모집 마감일이 다가오고 주변에서는 카테캠에 지원했다는 사람들의 소식이 들려오기 시작하면서 왜 해보지도 않고 포기해야 하지?라는 생각이 들어 안되더라도 해보고 탈락하는 게 마음이 편할 것 같아 지원해 보았다. 이러한 마음을 먹고 나서 집에 들어와 나는 곧바로 카테캠 홈페이지를 들어가 ..

JPA(Java Persistence API)란?자바를 사용하는 응용 프로그램에서 관계형 데이터베이스의 관리를 위한 인터페이스 및 기능을 제공하는 자바 API이다.(JPA에 대한 자세한 내용은 추후 다른 게시글 URL 첨부 예정) JPA의 특징: SQL 문장을 작성하지 않고도 JPA Query method naming을 통해 JPQL쿼리를 수행하여 데이터베이스를 관리할 수 있다.  > 이때, 메서드를 아무렇게나 작성하면 안 되고 정해진 규칙에 의해 작성해야 한다.: CrudRepository에 사전 정의된 메서드 사용 시, 따로 Repository interface에 메서드를 선언하지 않아도 사용할 수 있다. 쿼리 메서드를 조합하는 방법은 다음과 같다기본 동작 메서드 + 반환 데이터 + By + 검색 조..

HTTP란?HyperText Transfer Protocol의 약자로, 서버와 클라이언트 사이에 데이터를 어떻게 주고받을지 미리 정해놓은 규칙(프로토콜, protocol)이다. 클라이언트(웹 브라우저 등)는 사전에 정의된 HTTP에 따라 서버에게 요청을 보내고 서버는 마찬가지로 HTTP에 따라 클라이언트의 요청에 대한 응답을 보낸다. 이때, 서버가 보내는 응답 메시지에는 상태 코드(Status Code)를 포함하는데 이는 100~500번대까지 세자리 숫자로 백의 자릿수에 따라 총 5개의 클래스(분류)로 구분되고, 이는 RFC 9110에 정의되어 있으며 '인터넷 할당 번호 관리기관' IANA(Internet Assigned Numbers Authority)에서 관리가 이루어지고 있다.1XX: 정보 제공(I..