dev-lee 님의 블로그

Airflow Athena CTAS ETL DAG S3에 있는 CSV 데이터를 Athena로 조회하고, 90점 이상 학생만 필터링하여 Parquet 포맷으로 재저장하는 CTAS 기반 ETL 파이프라인전체 파이프라인 흐름[S3 Clean] → [Table Drop] → [CTAS 쿼리 실행] → [완료 감지]매 실행 시 기존 리소스를 초기화(멱등성 보장) 한 뒤, Athena의 CTAS(Create Table As Select)로 데이터를 필터링하여 Parquet 포맷으로 저장하는 구조.1. 모듈 임포트from datetime import datetime, timedeltafrom airflow import DAGfrom airflow.providers.amazon.aws.operators.athena i..

1. Novo Nordisk와 OpenAI 전략적 파트너십 체결 — AI로 신약 개발 가속화덴마크 제약사 Novo Nordisk와 OpenAI가 신약 개발 가속화와 글로벌 운영 전반의 AI 통합을 목표로 전략적 파트너십을 체결하였습니다. 2026년 말까지 AI를 회사 전반에 도입하는 것을 목표로 삼은 이번 협력은 빅파마와 프론티어 AI 기업 간의 대형 제휴 사례 중 하나로 주목받고 있습니다.Novo Nordisk는 비만 치료제 Wegovy, 당뇨 치료제 Ozempic으로 세계적인 인지도를 확보한 제약사입니다. 이번 파트너십을 통해 OpenAI의 최신 모델과 에이전트 기술을 활용하여 신약 후보 물질 발굴, 임상 데이터 분석, 의약품 공급망 최적화까지 전방위적으로 AI를 적용할 계획입니다.특히 신약 개발 영..

실시간 주가 집계 파이프라인 구축 프로듀서가 주가 로그를 KDS로 전송하면, Flink가 10초 단위로 종목별 평균가를 집계하여 출력 KDS로 보내고, Firehose가 이를 받아 S3에 적재하는 실시간 스트림 처리 파이프라인 구조전체 아키텍처kds_flink_adf_producer.py (로컬 프로듀서) ↓ put_record (1초 간격)KDS Input (de-ai-04-an2-kds-stock-input) ↓ Flink가 LATEST 포지션부터 구독Flink Application (de-ai-04-an2-flink-stock) ├─ stock_input 테이블 (KDS 입력 매핑) ├─ TUMBLE 윈도우 연산 (10초 단위 ticker별 AVG) └─ stock_o..

주가 로그 파이프라인 구축 — 프로듀서부터 KDS, Firehose, S3 적재까지boto3 프로듀서가 가짜 주가 로그를 Kinesis Data Streams로 전송하고, Firehose가 해당 스트림을 구독하여 S3에 배치 적재하는 실시간 파이프라인 구조전체 아키텍처stock_log_generator.py (로컬) ↓ put_record (1초 간격)KDS (de-ai-04-an2-kds-stock-input) ↓ 구독 (자동)Firehose (de-ai-30-an2-adf-stock-analysis) ↓ 버퍼링 후 배치 전송S3 (de-ai-30-827913617635-ap-northeast-2-an)로컬 프로듀서 → KDS → Firehose → S3로 이어지는..

run.py + adf_direct_data_put.py 코드 분석 — 로그 생성부터 Firehose 전송까지LogGenerator로 로그를 생성하고, Amazon Data Firehose를 통해 S3로 스트리밍하는 파이프라인 구조전체 아키텍처LogGenerator (log_generator.py) ↓ run.py ←── make_log() : 로컬 테스트용 (N건 반복 생성 + 콘솔 출력) │ └──── make_one_log() : 1건 생성 + JSON 문자열 반환 ↓ adf_direct_data_put.py ↓ Firehose (put_record) ..

LogGenerator 클래스 코드 분석 — 도메인별 로그 생성기Faker와 random을 조합하여 6개 도메인의 가상 로그를 생성하는 클래스. random.choices의 가중치 확률로 현실적인 이벤트 분포를 시뮬레이션하는 구조전체 구조LogGenerator├── get_interval_time() — 로그 발생 간격 계산 (fixed / random)├── finance() — 금융 거래 로그├── ecommerce() — 이커머스 행동 로그├── iot() — IoT 센서 로그├── ott() — OTT 스트리밍 로그├── factory() — 스마트팩토리 생산 로그└── lol_game(..

Airflow S3 Producer-Consumer DAG 코드 분석 — 이벤트 기반 파이프라인Producer가 S3에 데이터를 업로드하면 Consumer가 감지하여 처리 → 삭제하는 이벤트 드리븐 파이프라인 구조전체 아키텍처[Producer DAG] [Consumer DAG]CSV 생성 → S3 업로드 ──→ S3 감시(Sensor) → 읽기/처리 → 삭제 ↑ S3: income/sensor_data.csv두 DAG가 S3의 income/sensor_data.csv 경로를 매개로 연결됨. Producer가 파일을 올리면 Consumer의 센서가 감지하여 후속 처리를 수행하는 구조.Pa..

1. 스탠포드 2026 AI 인덱스 보고서 발표 — "AI는 전력 질주 중, 우리는 아직 신발을 찾는 중"스탠포드 대학교 인간 중심 AI 연구소(HAI)가 2026년 AI 인덱스 보고서를 발표했습니다. 400페이지가 넘는 이번 보고서는 벤치마크 성능부터 투자, 대중 인식까지 AI 산업 전반을 다각도로 분석하고 있습니다.가장 눈에 띄는 대목은 프론티어 모델들의 성능 향상 속도입니다. 전문가 수준의 난이도로 유명한 벤치마크 'Humanity's Last Exam'에서 2025년 최고 모델(OpenAI o1)은 정답률 8.8%에 불과했지만, 2026년 4월 현재 Anthropic의 Claude Opus 4.6과 Google의 Gemini 3.1 Pro는 50%를 넘어섰습니다. 박사 수준의 과학 문제, 멀티모달..

Airflow - AWS S3 연동 기초 (파일 업로드 DAG)Airflow에서 AWS S3에 파일을 업로드하고, 업로드 결과를 확인하는 기본적인 DAG를 분석해보았다. AWS 액세스 키가 정상 작동하는지 체크하고, 로컬 파일을 S3에 적재하는 흐름을 다루었다.전체 구조파일 생성 (Bash) → S3 업로드 → 업로드 확인BashOperator로 로컬에 텍스트 파일을 생성하고, LocalFilesystemToS3Operator로 S3에 업로드한 뒤, S3Hook으로 실제 업로드 여부를 검증하는 흐름.1. 모듈 가져오기from datetime import datetime, timedeltafrom airflow import DAGfrom airflow.operators.python import Python..

Airflow MSA 신용평가 DAG - 기존 버전 vs 업그레이드 버전 비교 분석MSA 환경에서 AI API 서버를 호출하여 고객 신용평가를 수행하는 Airflow DAG를 분석해보았다. 기존 버전과 업그레이드 버전의 차이점을 중심으로 코드를 하나하나 뜯어볼 예정이다.전체 구조 비교두 버전 모두 동일한 흐름을 가짐:더미 데이터 준비 → (데이터 추출) → AI API 호출 → 결과 DB 저장핵심 차이는 데이터를 어디서 가져오고, 어떻게 저장하느냐에 있다. 항목기존 버전업그레이드 버전데이터 소스파이썬 리스트(하드코딩)DB에서 SELECT 조회더미 데이터고정 3건랜덤 50건, DB에 INSERTExtract Task미구현 (pass)DB 조회로 구현결과 저장 방식INSERT (신규 삽입)UPDATE (기존..