Data

Airflow에서 DAG파일을 관리하고 배포하는 방식은 여러 가지가 있다. 이전에는 nfs를 pv로 활용해서 nfs스토리지에 DAG파일을 놓고 사용해봤고 더 효율적인 방법을 위해 gitsync를 설정해보자.1. SSH 키 생성 및 GitHub 등록Git 저장소에 안전하게 접근하기 위해 SSH 키 쌍을 생성한다.# SSH 키 생성 (RSA 4096bit)ssh-keygen -t rsa -b 4096 -C "your_email@example.com"생성 경로: /root/.ssh/id_rsa_airflowPassphrase: 입력하지 않고 엔터로 진행한다.생성된 공개키(id_rsa_airflow.pub)의 내용을 복사하여 GitHub Repository > Settings > Deploy keys에 등록한다..

Airflow 학습을 목적으로 Upbit API를 활용해 매시간 비트코인(KRW-BTC) 분봉 데이터를 수집하고 저장하는 파이프라인을 만들어보자. 데이터 저장소는 Kubernetes 환경에 Rook Ceph를 배포하여 구축한 S3 호환 Object Storage를 사용했다.데이터는 Raw 형태(JSON)로 먼저 저장한 후, 분석 효율을 위해 Parquet 포맷으로 변환하여 Processed 영역에 다시 저장한다.1. Airflow Connections 설정DAG 코드 내에서 인증 정보를 하드코딩하지 않고, Airflow의 Connection 기능을 활용해 안전하게 관리한다.1.1 S3 Connection (Rook Ceph)Rook Ceph로 생성한 S3 버킷에 접근하기 위한 설정이다. Connectio..

Rocky Linux 8.10이 설치된 가상머신(VM) 4대로 구성된 Kubernetes 클러스터 환경을 기준으로 Rook Ceph(v1.18.8)를 배포하고, Object Storage를 구성하는 과정을 기술한다. 각 VM은 Kubernetes 노드로 동작 중이며, Ceph 스토리지 구성을 위한 별도의 블록 디바이스가 할당된 상태를 전제로 한다.1. 사전 요구 사항 (Prerequisites)Rook Ceph를 설치하기 전, Rocky Linux 8.10 노드에서 스토리지 구성을 위한 사전 작업이 필수적이다. 특히 Ceph OSD(Object Storage Daemon)로 사용할 디스크는 어떠한 파일 시스템이나 파티션도 존재하지 않는 Raw Device 상태여야 한다. 나는 control-plane노드..

Kubernetes 환경에 배포된 Airflow에서 SSHOperator를 테스트하던 중, 파드(Pod)가 재시작되거나 작업이 종료된 후 로그가 사라지는 현상을 확인했다. 이를 해결하기 위해 로그 영구 저장(Persistence) 설정을 적용하는 과정과, 그 과정에서 발생한 airflow-triggerer StatefulSet 업데이트 에러를 해결하는 방법을 정리한다.1. SSHOperator 테스트 환경 구성먼저 외부 서버에 명령을 수행하기 위해 Admin Connection을 생성하고 DAG를 작성한다.1-1. Connection 생성 1-2. DAG 작성DAG PVC를 nfs로 지정했기 때문에 해당 nfs 위치에서 파일 생성 진행함. from datetime import datetimefrom ai..

VM 기반 쿠버네티스 클러스터 환경에서 외부 PostgreSQL을 연동하여 Airflow 3.0.2 버전을 배포하는 과정을 정리한 문서입니다.1. 설치 환경항목상세 정보비고PlatformKubernetes Cluster (v1.32.3)VMAirflow3.0.2Helm Chart v1.18.0DBPostgreSQL 16.10외부 DB (별도 설치) 2. Helm 차트 준비Airflow 공식 Helm 저장소를 추가하고, 설치하려는 Airflow 버전(3.0.2)에 해당하는 차트 버전을 확인한다.# Helm 레포지토리 추가helm repo add apache-airflow https://airflow.apache.org# 차트 버전 확인 (Airflow 3.0.2는 Chart 1.18.0에 해당)helm s..

1. replication.factor (복제 계수)역할: 하나의 토픽 파티션 데이터를 총 몇 개의 복제본으로 저장할지 결정한다.설정 위치: 브로커(서버) 측, 주로 토픽을 생성할 때 지정한다.구성: 계수가 N일 때 복제본은 1개의 리더(Leader)와 N-1개의 팔로워(Follower)로 구성된다.리더: 모든 읽기(Read)와 쓰기(Write) 요청을 처리한다.팔로워: 리더의 데이터를 그대로 복제한다. 평상시에는 요청을 처리하지 않다가, 리더 브로커에 장애가 발생하면 팔로워 중 하나가 새로운 리더로 선출된다.주의: replication.factor는 가용한 브로커 수보다 클 수 없다.2. min.insync.replicas (최소 동기화 복제본)min.insync.replicas는 "데이터를 보냈다고 ..

Mlops에 대해 공부해보고자 지난번에 Kubeflow를 구축했다.Kubeflow 설치: https://hd-engineering.tistory.com/23모델 레지스트리 설치: https://hd-engineering.tistory.com/24이번에는 테스트 코드를 활용하여 Kubeflow에서 학습을 진행하고, 학습된 모델을 모델 레지스트리에 저장한 뒤, 증분 학습을 시키는 과정을 수행해 보았다. pipeline.py 테스트 코드 설명실제로 의미 있는 모델을 만들면 좋겠지만.. 그것이 목적이 아니므로, 단순히 과정을 실습해보기 위한 테스트 코드를 AI를 활용해 생성했다.애플 주가 예측 모델을 예시로 사용했다.1년치 데이터를 외부에서 받아 학습하도록 구성했다.Kubeflow에서 년도와 기존 모델 경로를 ..

EKS 환경에서 MSA로 구축된 서비스 로그를 Fluent Bit을 사용하여 OpenSearch로 적재하고 있다. 이 과정에서 발생하는 시간 불일치 문제를 해결하고, 로그 내의 실제 시간을 OpenSearch의 기본 타임스탬프로 사용하는 방법을 정리한다.문제 상황: @timestamp와 실제 로그 시간의 불일치일반적으로 수집되는 서비스 로그의 예시는 다음과 같다.2025-09-17 10:16:14.223 32487 --- [nio-8080-exec-3] INFO o.s.web.servlet.DispatcherServlet : POST "/api/users/login", parameters={}2025-09-17 10:16:14.225 32487 --- [nio-8080-exec-3] DEB..