지난번에 AWS EC2를 띄어서 환경 셋팅까지 완료하였다.
이제 본격적으로 실습을 진행할 예정이다. 강의에서는 Kafka로 먼저 실습하는 과정이 있다.
Kahka란?
아파치 카프카(Apache Kafka)는 대용량의 실시간 데이터(로그, 이벤트 등)를 막힘없이 빠르게 주고받고 안전하게 저장해 주는 초고속 데이터 메시징/스트리밍 플랫폼이다.
여러 시스템 사이에서 데이터를 보내는 쪽과 받는 쪽을 분리하여, 데이터가 폭주해도 유실 없이 안정적으로 실시간 파이프라인 처리를 가능하게 해준다.
kafka Server 생성하기
kafka라는 이름으로 ec2 인스턴스 3개를 생성한다.
실습을 위한 SSH 접속
실습 과정에 웹서버 띄우기 과정이 있어서 해당 내용 진행
(보안그룹 80 포트 inbound 설정 필요)
꿀팁을 강사님이 주셨는데 아래와 같이 EC2 생성 시 '고급 세부 정보'를 통해
바로 웹서버가 실행되게 할 수 있다고 한다.
- web service 실행2- ec2 생성시
#!/bin/bash
yum update -y
yum install httpd -y
sudo service httpd start
echo "내용" > /var/www/html/index.html
Kafka 다운로드 받기
kafka_server 인스턴스에 접속해서 아래 깃허브에 있는 명령어로 다운받는다.
wget https://archive.apache.org/dist/kafka/3.6.1/kafka_2.13-3.6.1.tgz
*wget 명령어는 웹페이지 URL 다운로드하는 것을 cli로 하게 해주는것이다.
압축을 풀고 파일들을 확인한다.
주키퍼 실행하기
./bin/zookeeper-server-start.sh config/zookeeper.properties & (마지막에 & 표시는 백그라운드로 실행을 의미한다.)
아파치 주키퍼(ZooKeeper)는 여러 대로 묶여 돌아가는 카프카 브로커(서버)들이 서로 엉키지 않도록 상태를 조율하고 관리해 주는 중앙 관제탑(코디네이터) 역할을 하기 때문이다.
카프카(기존 버전 아키텍처 기준)는 자체적으로 브로커들의 상태나 메타데이터를 직접 동기화하지 않고, 주키퍼에게 이 역할을 맡기도록 설계되었습니다.
⚬ 브로커 상태 감시 (Health Check): 여러 카프카 브로커 중 어떤 서버가 살아있고 어떤 서버가 다운되었는지 실시간으로 모니터링한다.
⚬ 컨트롤러(Leader) 선출: 카프카 브로커들 중 전체 클러스터를 총괄할 대장 역할(컨트롤러)을 주키퍼를 통해 투표하고 선출한다.
⚬ 메타데이터 저장: 어떤 토픽이 존재하고, 각 파티션의 리더/팔로워가 누구인지 등의 핵심 설정 정보를 주키퍼에 안전하게 보관한다.
⚬ 토픽 및 ACL 변경 감지: 새로운 토픽이 생기거나 접근 권한 설정이 변경되면 이를 모든 브로커에 즉각 전파한다.
카프카 실행하기
./bin/kafka-server-start.sh config/server.properties &
카프카의 Topic이 무엇인가? 생성하기
토픽(Topic)은 카프카 안에서 데이터를 목적이나 주제별로 구분해 담아두는 데이터 저장소(폴더/게시판 채널)이다.
프로듀서(데이터를 보내는 쪽)와 컨슈머(데이터를 읽는 쪽)는 카프카 전체를 바라보는 것이 아니라, 특정 토픽을 지정해 데이터를 넣고 꺼내온다.
⚬ 주제별 데이터 분리: 예를 들어 order-events(주문 내역), user-logs(사용자 로그), payment-history(결제 내역)처럼 성격이 다른 데이터들을 토픽 단위로 나누어 관리한다.
⚬ 파티션(Partition) 구성: 하나의 토픽은 병렬 처리 성능을 높이기 위해 여러 개의 파티션으로 쪼갤 수 있습니다. 데이터가 들어오면 파티션들에 분산 적재되어 여러 컨슈머가 동시에 나눠 읽을 수 있다.
⚬ 다중 컨슈머 지원 (1:N 구독): 하나의 토픽에 들어온 메시지는 한 번 읽는다고 사라지는 것이 아니라, 서로 다른 여러 컨슈머 그룹(예: 분석 시스템, DB 적재 시스템)이 각자의 속도로 동시에 읽어갈 수 있다.
⚬ 데이터 보존 주기: 일반적인 메시지 큐와 달리 데이터를 읽어가도 즉시 삭제되지 않고, 설정된 보존 기간(기본 7일 등) 동안 디스크에 안전하게 유지된다.
아래 깃허브의 명령어를 통해 apartinfo라는 토픽을 생성한다.
bin/kafka-topics.sh --create --topic apartinfo --partitions 1 --replication-factor 1 --bootstrap-server localhost:9092 &
Producer 서버에 설치
./bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic apartinfo --from-beginning
kafka server애서 서비스 띄우고 kafka_producer ec2에
데이터(메시지)를 생성하여 카프카의 특정 토픽(Topic)으로 쏘아 보내는(발행하는) 발신자 애플리케이션을 만든다.
logstash를 통해 실습을 해보기
아래 Producer에서 logstash 설치한다.
sudo yum install logstash -
S3에서 데이터 가져오는 파이프라인 하나 생성
$ vi apartinfo_test.conf
input {
s3 {
sincedb_path => "/dev/null"
access_key_id => "accesskey"
secret_access_key => "security_key"
region => "ap-northeast-2"
prefix => "ods/danji_master.json/" #bucket 하위 폴더를 지정해준다.
bucket => "fc-storydata" #bucket 명을 지정해준다.
additional_settings => {
force_path_style => true
follow_redirects => false
}
}
}
output {
stdout { }
}
실행시킨다.
logstash -f /home/ec2-user/apartinfo_test.conf
kafka 서버로 전송이 되도록 config 수정!(카프카서버 private ip로 등록)
성공!!
내용이 많아서.. 이후 Consumer 부분 이어서 실습 예정이다.
Consumer 서버도 동일하게 SSH로 접근하고 아래 명령어를 실행한다.
logstash -f /home/ec2-user/apartinfo_test.conf
[카프카 데이터 파이프라인 실습 요약]
1. 파이프라인 구조
⚬ Source (S3): 원본 데이터(아파트 정보 JSON)가 저장된 저장소.
⚬ Producer (Logstash): S3의 데이터를 실시간으로 읽어와 카프카 토픽으로 전송.
⚬ Queue (Kafka): 데이터를 안전하게 적재 및 전달하는 메시지 큐(Topic).
⚬ Consumer : 카프카 토픽에 쌓인 데이터를 실시간으로 소비 및 출력.
2. 실습 목표
⚬ 데이터 생산자와 소비자를 분리하여 시스템의 유연성을 확보하고, 대용량 데이터를 안정적으로 처리하는 스트리밍 파이프라인의 기초를 체득함.
3. 데이터 흐름(결과물)
⚬ S3 → Logstash → Kafka(apartinfo 토픽) → Kafka Consumer
⚬ 최종 결과: S3에 있던 아파트 정보 데이터가 카프카를 거쳐 컨슈머 터미널에 실시간 JSON 포맷으로 출력되는 것을 확인.
4. 느낀 점
⚬ 데이터가 파일(배치) 상태가 아닌 흐르는 데이터(스트림) 형태로 전달되는 과정을 직접 구현하며, 카프카가 데이터 허브로서 왜 중요한지 이해할 수 있었음.
⚬ 추후 이 구조를 바탕으로 Elasticsearch나 DB로 적재하는 컨슈머를 붙여 검색 및 분석 서비스로 확장 가능함.
