데이터 파이프라인 실습

웹 로그를 생성하여 Kafka, MongoDB, MySQL을 거치는 데이터 파이프라인 실습 프로젝트

아키텍처

Log Generator --> Kafka --> MongoDB Consumer --> MongoDB (Raw)
                                                      |
                                                      v
                                            Batch Processor (pandas)
                                                      |
                                                      v
                                                   MySQL (정형화)

사전 요구사항

• Docker & Docker Compose
• Python 3.9+

실습 순서

1. 인프라 실행

docker-compose up -d

컨테이너 상태 확인:

docker-compose ps

모든 컨테이너가 running 상태가 될 때까지 대기 (약 30초)

2. UI 도구 접속

서비스	URL	설명
Kafka UI	http://localhost:8080	Kafka 토픽, 메시지, 컨슈머 그룹 모니터링
Mongo Express	http://localhost:8081	MongoDB 데이터 조회 및 관리
Adminer	http://localhost:8082	MySQL 데이터 조회 및 관리

Adminer 접속 정보:

• 시스템: MySQL
• 서버: mysql
• 사용자: root
• 비밀번호: password
• 데이터베이스: pipeline_db

3. Python 의존성 설치

pip install -r requirements.txt

4. 로그 생성기 실행

터미널 1에서 실행:

python generator/log_generator.py

0.5초마다 가상 웹 로그가 생성되어 Kafka로 전송됨

5. MongoDB Consumer 실행

터미널 2에서 실행:

python consumer/mongodb_consumer.py

Kafka에서 메시지를 소비하여 MongoDB에 저장

6. 데이터 확인 (CLI)

MongoDB 접속하여 데이터 확인:

docker exec -it boaz_mongodb mongosh -u root -p password

use pipeline_db
db.raw_logs.find().limit(5)
db.raw_logs.countDocuments()

7. 배치 처리 실행

터미널 3에서 실행:

python batch/batch_processor.py

MongoDB의 미처리 데이터를 읽어 MySQL에 정형화하여 저장

8. MySQL 결과 확인 (CLI)

docker exec -it boaz_mysql mysql -u root -ppassword pipeline_db

-- 이벤트 조회
SELECT * FROM events LIMIT 10;
SELECT COUNT(*) FROM events;

-- 일별 사용자 통계
SELECT * FROM daily_user_stats ORDER BY stat_date DESC LIMIT 10;

-- 시간별 페이지 통계
SELECT * FROM hourly_page_stats ORDER BY stat_datetime DESC LIMIT 10;

-- 배치 처리 기록
SELECT * FROM batch_status ORDER BY start_time DESC LIMIT 5;

프로젝트 구조

BOAZ-Data-Pipeline/
├── docker-compose.yml          # 인프라 정의
├── requirements.txt            # Python 의존성
├── README.md                   # 실습 가이드
├── config/
│   └── config.yaml             # 설정 파일
├── schemas/
│   └── mysql_init.sql          # MySQL 테이블 스키마
├── generator/
│   └── log_generator.py        # 로그 생성기
├── consumer/
│   └── mongodb_consumer.py     # Kafka Consumer
└── batch/
    └── batch_processor.py      # 배치 처리기

생성되는 데이터

Raw 로그 (MongoDB)

{
    "event_id": "uuid",
    "timestamp": "2025-11-26T14:30:00.123Z",
    "user_id": "user_0001",
    "session_id": "sess_user_0001_1",
    "event_type": "page_view",
    "page_url": "/products/1",
    "device": "mobile",
    "browser": "Chrome",
    "ip_address": "192.168.1.100",
    "metadata": {}
}

정형화된 테이블 (MySQL)

• events: 이벤트 원본 정형화
• daily_user_stats: 일별 사용자 통계 (page_views, clicks, purchases, total_spent)
• hourly_page_stats: 시간별 페이지 통계 (view_count, unique_users)
• batch_status: 배치 처리 이력

종료

# 모든 컨테이너 종료 및 볼륨 삭제
docker-compose down -v