일단 지난 3, 4월부터 해서 오늘까지 꽤 시간이 많이 지났습니다. 그 안에 많은 상황들이 있었고, 제가 목표했던 것들을 조정하기도 이뤄내기도 했습니다. 일단 저번에 올렸던 목표를 점검해보겠습니다. 알고리즘 스터디 CS 스터디(정처기 실기 준비)(가채점 결과는 합격) 빅데이터를 지탱하는 기술 책 리뷰(읽기만 했습니다,,) 데이터엔지니어링 키트 마무리 블로그 리뷰하기 일단 알고리즘 스터디는 전혀 하지 못했고, 정처기 실기 시험은 잘 마무리가 됐습니다. 그리고 프로그래머스에서 진행하는 데이터엔지니어링 키트 모두 완강했고, 공부하면서 블로그 리뷰도 마무리 했습니다. 사실 5월 달까지, 전부하진 못할거라고 생각했는데 그래도 중요한 우선 순위를 잘 두고 하나하나 잘 했던 것 같습니다. 마냥 데이터가 좋아서 이 길..
SPARK 버클리 대학의 AMPLab에서 아파치 오픈소스 프로젝트로 2013년 시작한 프레임워크입니다. 하둡의 뒤를 잇는 2세대 빅데이터 기술로, Yarn(Yarn은 Hadoop 2.0, 3.0을 말한다고 보시면 됩니다.)을 분산환경으로 사용하고 Scala로 작성되었습니다. 일단 Spark의 기본적인 개념은 분산, 병렬 처리입니다. 데이터를 나누고 이 각각, Block(Partition)을 따로 가공 및 정제 후에 합치는 방식입니다. Spark은 하둡의 Mapreduce 방식과 다르게 디스크에 저장하는 방식이 아닌, 메모리에 저장함으로써 속도를 월등히 높였습니다. 특히 MapReduce는 프로그래밍에서 사용할 수 있는 Operator가 딱 두 개였고, SQL로 치면 GROUP BY, CASE WHEN 정..
Flume(with Kafka) Flume은 대규모 데이터 처리를 위한 분산형 데이터 수집 도구입니다. Flume은 데이터를 수집하고 처리하여 저장소 또는 전송 수단으로 전송합니다. Flume은 다양한 데이터 소스 (예 : 로그 파일, 소켓 등)에서 데이터를 수집하고, 이를 다양한 대상 (예 : HDFS, Kafka 등)으로 전송할 수 있습니다. Flume은 이러한 과정에서 데이터를 중간 단계에서 필터링하거나 변환할 수도 있습니다. Flume은 Kafka와 자주 사용됩니다. Kafka는 분산형 스트리밍 플랫폼으로, 대용량 실시간 데이터 스트리밍 및 처리에 사용됩니다. Kafka는 대규모의 메시지를 안정적이고 신뢰성 높은 방식으로 처리하며, 다양한 데이터 소스에서 데이터를 수집하고 다양한 대상으로 전송할 ..
Dag간 실행순서를 정하기 위해서는 Dag Dependencies를 활용하면 됩니다. 즉 어떤 Dag가 실행되면, 따라오는 Dag가 있게끔 설정하는 것이 가능합니다. 보통 TriggerDagOperator를 쓰는데, ExternalTaskSensor는 성능 상의 이유로 잘 쓰지 않습니다. 이유는 후술하겠습니다! Explicit Trigger 이 Trigger 방법은 한 Dag는 다음 Dag를 트리거 시킵니다. 한 방법으로, DAG A의 태스크를 TriggerDagRunOperator로 구현하는 방법입니다. 트리거 A는 다음과 같이 구현하면 됩니다. 이때 유의할 점은 반드시 airflow.cfg dag_run_conf_overrides_params가 True로 설정 되어 있어야합니다. from airflo..
Airflow는 환경설정을 통해서, 웹 api를 외부에서 활용할 수 있습니다. Airflow API 활성화 airflow.cfg의 api 섹션에서 auth_backend의 값을 변경합니다. auth_backend = airflow.api.auth.backend.basic_auth airflow 웹서버를 재실행합니다. sudo systemctl restart airflow-webserver basic_auth 설정 재확인 sudo su airflow airflow config get-value api auth_backends Airflow Web UI 에서 새로운 사용자 추가 Security -> List Users -> + 이후 화면에서 새 사용자 정보를 추가 합니다(monitor:MonitorUser1..
슬랙 메시지를 활용해서 데이터파이프라인을 효율적으로 관리할 수 있습니다. 보통 현업에 계신 분들은 데이터파이프라인 에러를 슬랙으로 받고, 해당 메시지를 받으면 데이터 파이프라인을 backfill 혹은 복구하는 작업을 합니다. 이번엔 airflow와 슬랙을 연동해서 에러메세지를 받아보는 작업을 해보겠습니다. 1.Incoming Webhooks App 생성하기 https://api.slack.com/messaging/webhooks Sending messages using Incoming Webhooks Creating an Incoming Webhook gives you a unique URL to which you send a JSON payload with the message text and some..
에어플로우 환경 설정 관련 고려사항 airflow.cfg 에어플로우의 환경설정 파일입니다. /var/lib/airflow/airflow.cfg 해당 경로에 위치합니다. 환경설정의 변경은 웹 서버 및 스케줄러를 다시 시작할 때 반영됩니다. dag_dir_list_interval : dags_folder를 Airflow가 얼마나 자주 스캔하는지 명시합니다(초 단위, default : 300) DAGs 폴더 [core] 섹션의 dags_folder가 반드시 DAG들이 있는 디렉토리가 되어야합니다. Airflow Database 에어플로우의 메타데이터 데이터베이스는 기본적으로 sqlite을 사용합니다. 다만 추후 확장을 위해 보통은 postgresql로 사용합니다. 이 데이터베이스는 주기적으로 백업되어야하며, ..
DOCKER DOCKER는 컴퓨터 안의 컴퓨터라고 생각하시면 좋습니다. 기존의 가상환경을 구축하는 것보다, 더 경량화된 컨테이너를 구축하는 작업이라고 보시면 됩니다. 다만 운영체제의 사양이 어느정도 받쳐줘야 사용할 수 있습니다. 특히 AIRFLOW의 경우에는 DOCKER CONTAINER에 8기가 정도의 메모리를 줘야 에어플로우를 쓰는데 문제가 없습니다. DOCKER IMAGE 단순히 응용 프로그램 뿐만 아니라 그 프로그램이 필요로 하는 모든 다른 환경까지 포함한 소프트웨어 패키지 Docker Registry에 가면 다양한 Docker Image들을 찾아볼 수 있습니다. Docker Container Docker Image를 Docker Engine에서 실행한 것을 지칭 Docker Engine만 실행하..
이번 과제는 NPS를 구해서, 해당 NPS를 단 한번만 업데이트하는 ELT 구현입니다. NPS는 사용자의 서비스 추천정도를 물은 뒤에 0~10점으로 환산하여, 10점 혹은 9점을 준 고객의 비율에서 0~6점을 준 고객의 비율을 빼는 방법입니다. 간단하게 ELT를 구현하고 테스트 코드도 동시에 삽입해서 ELT 과정을 구현해보겠습니다. SQL 구현하기 저번 포스팅에서 만들어둔 nps 테이블을 활용하여 다음 SQL을 준비합니다. SELECT DATE(created_at) AS date, CAST(CAST(COUNT(CASE WHEN score IN (9,10) THEN 1 END) - COUNT(CASE WHEN score IN (0,1,2,3,4,5,6) THEN 1 END) as FLOAT) / COUNT..
BACKFILL 데이터파이프라인을 설계하고 운용할 때, 가끔 에러가 나거나, 이미 지난 날짜를 기준으로 데이터를 재처리 해야할 때가 있습니다. 이 때 백필은 재처리 작업을 의미합니다. 단어 의미 그대로 '매우는 작업'이라고 보시면 됩니다. Full Refrest를 한다면 backfill은 필요 없습니다. backfill은 일별 혹은 시간별 업데이트를 의미합니다. 마지막 업데이트 시간 기준 backfill을 하는 경우라면(데이터 웨어하우스 기록 시간 기준), execution_date을 이용한 backfill은 필요하지 않습니다. 데이터의 크기가 커질수록 backfill 기능을 구현해두는 것은 필수입니다. airflow는 이 backfill 작업을 굉장히 쉽게 만들지만, 데이터소스의 도움없인 불가능합니다...
5주차에 배운 내용들을 포스팅합니다. 먼저 OLTP와 OLAP의 차이에 대해 간단히 알아보도록 하겠습니다. OLTP VS OLAP OLTP(ONLINE TRASACTION PROCESSING) 온라인 트랜잭션 프로세싱, 즉 온라인에서 일어나는 데이터 교환에 중점을 두는 작업입니다. 데이터 처리 속도가 보다 중요하며, 큰 데이터를 다루기에는 적합하지 않습니다. 우리가 흔히 알고 있는 MYSQL, POSTGRESQL 같은 데이터베이스들이 이 작업에 중점을 둔 DB입니다. OLAP(ONLINE ANALYTIC PROCESSING) 온라인 애널리틱 프로세싱, 즉 분석을 중점을 두는 작업입니다. 데이터 처리 속도는 조금 떨어지더라도, 큰 과거 데이터를 활용해서 분석처리에 사용합니다. AWS REDSHIFT, GO..
Incremental Update의 경우에는 backfill 이슈가 발생하면서, 유지보수 비용이 full refresh에 비해 기하급수적으로 올라갑니다. 하지만, 데이터가 커질수록 full refresh의 경우에도 비용이 증가하면서, 오히려 Incremental Update가 나은 상황이 있을 수 있습니다. 따라서 해당 포스팅에서는 Incremental Update를 기초적으로 어떻게하는지 알아보겠습니다. 먼저 해당 포스팅에 대한 기본적인 정보는 앞선 포스팅을 참고해주세요. https://dataengineerstudy.tistory.com/161 [4주차] AIRFLOW 활용 데이터 적재(FULL REPRESH) FULL REFRESH 상황 가정 데이터 적재 FULL REFRESH란 테이블에 있는 모든..
