엔지니어 블로그

이전에 생성해둔 api 서버를 호출해서 데이터를 PubSub에 Publish 하는 작업을 할 차례다. 필요 작업은 다음과 같다.1.Airflow Connection 등록-> Airflwo가 PubSub에 접근할 수 있도록 Connection을 등록2.Publish Task 작성-> DAG에 데이터를 Publish 할 수 있도록 task 작성 1.Airflow Connection 등록먼저 connection 등록이다. Airflow 상단 메뉴에 Admin-Connection을 클릭하면 Connection 등록이 가능하다. + 버튼을 눌러 등록해보자 + 버튼을 눌러보면 다음과 같은 화면이 나온다. 여기서 필수로 작성해야 하는 것은 * 처리 되어있다. 추가로 Keyfile 내용을 작성하면 된다. Keyfile..

프로젝트를 하던 중 지금 토이 프로젝트를 하고있지만, 현업에서 진행하는 프로젝트라면 지속적으로 쌓이는 데이터의 양이 엄청날 것 같았다. 그래서 현재 Spark에서 Bigquery로 데이터를 전송하는 부분을 손보게 되었다.  이전 아키텍처와 거의 동일하고, 중간에 Cloud Storage가 들어갔다. 하단의 Base Data Load 부분은 정적 데이터의 적재를 뜻한다. 실시간으로 생성되는 데이터가 아닌, 현업이라면 DB에 이미 존재했어야 할 유저 정보, 지리정보 등의 데이터가 포함된다. 이 데이터를 업로드 하는 부분에서 고민이 생겼다. 현업에서 처리한다면 어떻게 했어야할까??Static한 데이터기 때문에 변화가 자주 없는 탓에 GCP 콘솔에서 직접 업로드 할까 생각했지만, 현업이라는 상황을 가정해봤을 때..

인프라 구성을 마친 후 데이터 Source가 되는 API 서버를 구축하기로 했다. 원래 DB에 저장 후 가져오는 방법을 사용하려고 했으나 그 방법은 자주 접해본 방법이기 때문에 API 서버를 직접 구축하고 데이터를 수집하는 방식으로 결정했다.  FastAPI를 사용하고 Postgresql과 연동하여 API를 구축했다. GET 메서드만 사용하는 간단한 서버이기 때문에 크게 어려움은 없었다. 고민한 지점은 한가지가 있었다.준비 된 데이터를 수집하는 주기는 어느정도로 잡아야 하는가?주기를 잡은 후에 어떤 기준으로 데이터를 가져 올 것인가?기존에 구상했던 것은 실제 웹 서버가 있고 사용자 행동 로그가 발생하는 것과 같은 방식이었다. 하지만 기준으로 발생 기준으로 잡을 컬럼이 애매했다. 그래서 ID 값을 따로 만..

아키텍처가 나온 후 인프라 구성에 들어갔다. 이전 글에 적어 둔 것처럼 k3s위에 airflow를 올려서 사용 할 예정이다.k3s의 설치는 아주 간단하다. 내부적으로 설정을 잡아 줄 일이 있다면 복잡해지지만, 단순히 설치만 하고 사용 할 것이라면 설치 스크립트 하나로 설치가 가능하다. Executor vs PodOperatorkubernetes에 airflow를 구성할 때 2가지 방식을 고려해 볼 수 있다. 제목에 있는 것 처럼 KubernetesExecutor와 KubernetesPodOperator를 사용하는 것이다. 두 방식의 차이는 결국 executor와 operator의 차이다. Executor는 task의 실행 방법을 결정한다. 대표적인 설정은 Celery,Local,Kubernetes등이 있..

최근 데이터 엔지니어링에 대해 다시 공부하고있다. 지금까지 공부한 내용들을 바탕으로 하나의 작은 프로젝트를 해보려고 한다. 그리고 프로젝트가 아닌, 프로젝트 진행 과정에서 나의 작업과 활동을 중심으로 글을 써 내려갈 생각이다. 진행기간2025년 2월 10일 ~ 2025년 2월 17일(예상) 프로젝트 목표Data Pipeline 구축본 프로젝트를 진행하는 목표는 지금까지 공부한 데이터 엔지니어링 생애 주기를 구현해보면서 실제 기술들이 어떤 상황에서 왜 사용되는지 직접 확인해보는 것이다. Infra 구축데이터 파이프라인이 구성 될 인프라 부터 구축 하는 것이 목표다. 프로젝트를 진행하며 파이프라인/인프라 구축 목표를 잡았다. 구축을 통해 나의 기술적인 지식들을 향상시키고 점검 할 수 있을 것이며 장애 대응 ..

Elasticsearch로 검색기능 구현 -4 에서는 각 필드에 알맞은 분석기를 적용해 원하는 형태의 토큰이 나오는 것을 확인했다. mapping과 indexing이 끝났으니 검색을 위한 search template을 구성할 것이다. search template란 검색이 유용하도록 미리 검색 형태를 지정해 두는 기능이다. 미리 query를 정해두고 필요할 때 마다 검색어만 입력하면 검색이 가능하다. search template을 정의하기 위해서는 다음과 같은 형태로 시작하게 된다. test_template 부분이 template의 이름이 된다. 이후로는 일반적인 쿼리와 동일하게 작성해 주게 되는데, 일반적인 쿼리와의 차이점은 검색어의 입력이다. 일반적으로 query를 사용해서 검색할 때는 "field" ..

지난 Elasticsearch로 검색기능 구현 -3 에서는 각 field를 원하는 type으로 mapping 하는 작업까지 완료했다. 이제는 각 field를 검색 조건으로 사용하기 위해 적절한 analyzer를 적용해볼 것이다. 1. 주소 analyze 우선, 원하는 검색의 형태는 다음과 같다. 전체 주소로 검색했을 때 검색이 가능한가? "서울시 봉익동" 으로 검색했을 때 검색이 가능한가 ? "서울시 종로구 봉익동" 으로 검색했을 때 검색이 가능한가 ? "종로구 봉익동" 으로 검색했을 때 검색이 가능한가 ? "봉익동" 으로 검색했을 때 검색이 가능한가 ? "서울 봉익동" 으로 검색했을 때 검색이 가능한가? 위의 검색 경우의 수를 모두 만족시키기 위해 nori analyzer를 활용할 것이다. "addr"..

더보기 1. mapping type 정하기 데이터의 전처리는 끝났고 Ls로도 출력을 해봤으니 Es에 indexing을 할 차례다. indexing 하기 전, index template를 사용하기 위해 미리 field의 mapping을 지정해 줘야 한다. 각 filed를 살펴보면 ["name","site_nm","middle_cate","detail_cate","cate","addr","load_addr","floor","date","timestamp"] 이렇게 구성되어 있다. 먼저 site_nm(지점명) 은 keyword type으로 지정해서 filter context를 사용할 것이다. cate 항목들과 addr(주소), name(상호명)은 text type으로 지정하고, date,timestamp는 d..