카프카, 데이터 플랫폼의 최강자 : 실시간 비동기 스트리밍 솔루션 Kafka의 기본부터 확장 응용까지 (1)

들어가며

기술적인 흐름/방향

• 느슨한 결합
• 비동기 메시징
• 강력한 메시지 처리 성능, 빠른 수평확장성, 고장감내성 -> 데이터 중앙화로 인한 데이터 분석
• 데이터 모니터링 / 데이터가 필요한 조직에 즉시 활용

---

1부 카프카를 시작하며

카프카의 탄생 배경, 기본적인 카프카의 개념과 특징, 클러스터 구성 설치 가이드

---

1장 카프카란 무엇인가

1.1 카프카의 탄생배경

대용량 메시지 데이터를 빠르게 처리하도록 개발된 메시징 플랫폼, 링크드인에서 출발해 2011년 초 아파치 공식 오픈소스로 공개

빅데이터를 분석할 때 여러 스토리지와 분석 시스템에 데이터를 연결하기 위한 필수 도구로 인식

기존 end-to-end 연결 방식의 아키텍처의 문제점

• 실시간 트랜잭션(OLTP) 처리와 비동기 처리가 동시에 이뤄지지만 통합된 전송 영역이 없으니 복잡도가 증가
• 각 파이프라인 별로 포맷과 처리법 이 달라 통합 관리(확장/변경/신뢰도)의 어려움

해결책 : 모든 시스템으로 데이터를 전송할 수 있고, 실시간 처리, 확장이 용이한 시스템

• 프로듀서와 컨슈머의 분리
• 메시징 시스템과 같이 영구 메시지 데이터를 여러 컨슈머에게 허용
• 높은 처리량을 위한 메시지 최적화
• 데이터가 증가함에 따라 스케일아웃이 가능

 

1.2 카프카의 동작 방식과 원리

중앙에 메시징 시스템 서버를 두고 메시지를 보내고 받는 Pub/Sub 모델

장점

• 받는쪽에 장애가 생겨도 메시지 유실X
• 메시징 시스템으로만 통신하기 때문에 확장 용이
• 교환기의 룰에 의해 데이터가 전달되므로 데이터 유실X

단점

• 직접 통신하지 않기 때문에 메시지 전달을 확인하려면 코드가 복잡해짐
• 중간에 메시징 시스템을 추가하므로 전달 속도 저하
• 보관,교환,전달 과정의 신뢰성에 중점 속도와 용량은 신경쓰지 않았기 때문에 간단한 이벤트를 서버로 전송하는데 주로 사용

카프카 성능 해결

• 교환,전달의 신뢰성 관리는 프로듀서와 컨슈머쪽에 전달
• 부하가 많이 걸리는 교환기 기능을 컨슈머가 만들 수 있도록

 

1.3 카프카의 특징

프로듀서와 컨슈머의 분리

• 구조를 단순화 하여 확장이 용이 해진다
• 한쪽 시스템에서 문제가 발생하더라고 연쇄작용이 발생할 확률이 매우 낮아진다

멀티 프로듀서, 멀티 컨슈머

• 하나의 토픽에 여러 프로듀서,컨슈머가 접근 가능
• 데이터 분석 및 처리 프로세스에서 하나의 데이터를 다양한 용도로 사용
• 중앙 집중형 구조

디스크에 메시지 저장

• 기존의 메시징 시스템과의 차이로 메시지가 컨슘되더라도 보관 주기 동안 디스크에 메시지를 저장
• 장점
  • 멀티 프로듀서, 멀티 커슈머 구조가 가능
  • 컨슈머 중지 후 실행되더라도 메시지 손실이 없다

확장성

• 하나의 카프카 클러스터는 3대의 브로커로 시작해 수십 대의 브로커로 확장 가능(서비스 중단 없이 가능)

높은 성능

• 내부적으로 분산 처리, 배치 처리 등 다양한 기법으로 고성능

 

1.4 카프카의 확장과 발전

카프카가 강력한 성능을 기반으로 일반적인 메시징 플랫폼을 넘어서

중앙 집중적 데이터 관리, 실시간 분석까지 가능한 데이터 파이프라인 플랫폼으로 진화 중이다

---

 

2장 카프카 설치

단순 설치 방법 외에도 Systemd를 이용한 프로세스 관리 방법과 카프카 상태를 확인하는 방법. 직접 설치 방법 (도커 설치는 부록에)

프로듀서 : 메시지를 보내는 역할

컨슈머 : 메시지를 가져오는 역할, 현재는 카프카 오프셋 정보를 주키퍼와 카프카 양쪽에 저장할 수 있지만, 향후 주키퍼에 오프셋을 저장하는 기능은 사라질 예정

주키퍼 : 컨슈머와 통신하는 부분 외에도 카프카와 직접 통신하여 카프카의 메타데이터 정보를 주키퍼에 저장하고, 카프카의 상태관리를 함, 카프카 패키지에 포함되어 있는 주키퍼를 사용하거나 별개로 주키퍼 웹사이트에 접속해 내려받아 사용 가능

2.1 카프카 관리를 위한 주키퍼

부하 분산 및 확장이 용이한 분산 애플리케이션 사용시 추가 개발 리소스를 줄이기 위해 이미 검증된 코디네이션 애플리케이션으로 주기퍼를 사용

분산 애플리케이션이 안정적인 서비스를 할 수 있도록 분산되어 있는 각 애플리케이션의 정보를 중앙에 집중하고 구성 관리, 그룹 관리 네이밍, 동기화 등의 서비스를 제공

• 서버 여러 대를 앙상블(클러스터)로 구성
• 분산 애플리케이션들이 각각 클라이언트가 되어 주키퍼 서버들과 커넥션을 맺은 후 상태 정보를 주고 받음
• 상태 정보들은 주키퍼의 지노드라 불리는 곳에 key-value 형태로 저장
• 지노드에 key-value가 저장된 것을 이용하여 분산 애플리케이션들은 서로 데이터를 주고 받음
• 지노드는 파일 시스템의 개념
  • 계층형 구조 (/app1/p_1/...)
• 데이터 변경 시 버전 번호를 증가하여 저장
• 메모리에 저장하여 처리량이 빠르다
• 앙상블로 구성되어 있는 주키퍼는 과반수 방식에 따라 살아 있는 노드 수가 과반수 이상 유지된다면 지속적인 서비스 가능
  • 주로 3개 이상 홀수로 구성
  • 많은 앙상블 구성시 대용량 처리에서 성능 향상

 

2.2 주키퍼 설치

• 자바가 설치돼 있어야 함
  • CentOS : yum -y install java-1.8.0-openjdk
  • Ubuntu : apt-get install openjdk-8-jdk
• 주키퍼 다운로드
  • zookeeper.apache.org 이동
  • Getting Started - Download 클릭
  • Download - Download 클릭 (아파치 미러 사이트 이동)
  • 가장 위에 있는 링크 (http://apachee.mirror.cdnetworks.com/zookeeper/)
  • 버전 확인
  • 원하는 디렉토리에서
    • wget http://apache.mirror.cdnetworks.com/zookeeper/zookeeper-3.4.10/zookeeper-3.4.10.tar.gz
  • 압축해제
    • tar zxf zookeeper-3.4.10.tar.gz
• 주키퍼 설정
  • 심볼릭 링크 생성 (버전 변경시에도 스크립트 수정 안하도록)
    • ln -s zookeeper-3.4.10 zookeeper
  • 지노드 복사본 스냅샷과 트랜잭션 로그들을 저장 할 경로 분리
    • mkdir -p /data
  • 앙상블 내 주키퍼 노드를 구분하기 위한 ID 생성 (주키퍼 설정 파일에서 사용됨)
    • echo 1 > /data/myid
    • echo 2 > /data/myid (주키퍼 수 만큼 반복)
  • 주키퍼 환경설정 파일 생성
    • vi /zookeeper경로/conf/zoo.cfg

    • tickTime=2000 #주키퍼가 사용하는 시간에 대한 기본 측정 단위
      initLimit=10  #팔로워가 리더와 초기에 연결하는 시간에 대한 타임 아웃 tick의 수
      syncLimit=5  #팔로워가 리더와 동기화 하는 시간에 대한 타임 아웃 tick의 수(주키퍼에 저장된 데이터가 크면 수를 늘려야 함)
      dataDir=/data  #주키퍼의 트랜잭션 로그와 스냅샷이 저장되는 데이터 저장 경로
      clientPort=2181  #주키퍼 사용 TCP포트
      server.1=peter-zk001:2888:3888  #주키퍼 앙상블 구성을 위한 서버 설정, server.myid 형식으로 사용
      server.2=peter-zk002:2888:3888
      server.3=peter-zk003:2888:3888

    • 더 많은 설정 정보
      • http://zookeeper.apache.org/doc/current/zookeeperAdmin.html#sc_configuration
• 주키퍼 실행
  • /zookeeper경로/bin/zkServer.sh start
  • /zookeeper경로/bin/zkServer.sh stop 종료
• systemd등록으로 효율적인 프로세스 관리 (CentOS 기준)
  • vi /etc/systemd/system/zookeeper-server.service

  • [Unit] #일반 옵션
    Description=zookeeper-server #해당 유닛에 대한 상세한 설명
    After=network.target # 유닛이 시작되는 순서를 조정 Before(지정된 유닛 실행 이전), After(지정된 유닛 실행 이후)
    ​
    [Service] #서비스 실행과 관련된 옵션
    Type=forking #EcexStart에 영향을 주는 유닛 프로세스가 시작
    User=root
    Group=root
    SyslogIdentifier=zookeeper-server #syslog에서 구분하기 위한 이름
    WorkingDirectory=/zookeeper경로 #실행된 프로세스의 작업 디렉토리 설정
    Restart=always #systemctl 명령어로 인한 중지를 제외하고 프로세스가 종료되면 재시작
    RestartSec=0s #몇초에 실행할지
    EcexStart=/zookeeper경로/bin/zkServer.sh start #서비스 시작될 때 실행할 명령어
    EcexStop=/zookeeper경로/bin/zkServer.sh stop #서비스 정지될 때 실행할 명령어

  • systemd 재시작
    • systemctl daemon-reload
• 주키퍼 실행
  • systemctl start zookeeper-server.service
  • systemctl stop zookeeper-server.service 중지
  • systemctl restart zookeeper-server.service 재시작
• 주키퍼 자동 실행 될 수 있도록 허용
  • systemctl enable zookeeper-server.service
• 상태 확인
  • systemctl status zookeeper-server.service

 

2.3 카프카 설치

클러스터 구성하는 서버의 대수를 정해야 함(자유롭게, 주키퍼 서버와 별도의 서버로 구성)

풀 GC 타임 동안 세션 타임 아웃 될 수 있으므로, GC주기를 체크하고 타임아웃도 3초 이상으로 설정

• 카프카 다운로드
  • 자바 설치
  • http://kafka.apache.org/downloads 이동
  • 원하는 디렉토리에서
    • wget http://apache.mirror.cdnetworks.com/kafka/1.0.0/kafka_2.11-1.0.0.tgz
  • 압축해제
    • tar zxf kafka_2.11-1.0.0.tgz
  • 심볼릭 링크 생성
    • ln -s kafka_2.11-1.0.0 kafka
• 카프카 설정
  • 디렉토리 준비
    • mkdir -p /data1
    • mkdir -p /data2
  • 카프카가 바라보는 주키퍼 설정
    • 주키퍼 앙상블 서버 리스트를 모두 입력 -> 주키퍼 지노드의 최상위 경로를 사용하여 하나의 주키퍼 앙상블이 하나의 애플레이케이션만 사용(데이터 충돌 방지)
      • zookeeper.connect=peter-zk001:2181,peter-zk002:2181,peter-zk003:2181
    • 주키퍼의 지노드 구분하여 사용 -> 하나의 앙상블이 여러개의 애플리케이션 사용
      • zookeeper.connet=peter-zk001:2181,peter-zk002:2181,peter-zk003:2181/peter-kafka01
      • zookeeper.connet=peter-zk001:2181,peter-zk002:2181,peter-zk003:2181/peter-kafka02
    • 카프카 환경설정 파일 수정
      • vi /kafka경로/config/server.properties
        • 브로커 아이디 변경
          • broker.id=1
        • 디렉토리 설정
          • log.dirs=/data1,/data2
        • 주키퍼 정보 수정
          • zookeeper.connet=peter-zk001:2181,peter-zk002:2181,peter-zk003:2181/peter-kafka
        • 파티션 수, 로그 보관주기,세션타임아웃 등 설정 가능
          • 자세한 설정 : https://kafka.apache.org/documentation/#brokerconfigs
• 카프카 실행
  • 옵션을 주어 실행
    • /kafka경로/bin/kafka-server-start.sh /kafka경로/config/server.properties -daemon
  • 중지
    • /kafka경로/bin/kafka-server-stop.sh
• systemd등록
  • vi /etc/systemd/system/kafka-server.service

  • [Unit] #일반 옵션
    Description=kafka-server #해당 유닛에 대한 상세한 설명
    After=network.target # 유닛이 시작되는 순서를 조정 Before(지정된 유닛 실행 이전), After(지정된 유닛 실행 이후)
    ​
    [Service] #서비스 실행과 관련된 옵션
    Type=simple #EcexStart에 영향을 주는 유닛 프로세스가 시작
    User=root
    Group=root
    SyslogIdentifier=kafka-server #syslog에서 구분하기 위한 이름
    WorkingDirectory=/kafka경로 #실행된 프로세스의 작업 디렉토리 설정
    Restart=no #systemctl 재시작 하지 않음
    RestartSec=0s #몇초에 실행할지
    EcexStart=/kafka경로/bin/kafka-server-start.sh #서비스 시작될 때 실행할 명령어
    EcexStop=/kafka경로/bin/kafka-server-stop.sh #서비스 정지될 때 실행할 명령어

  • 재시작
    • systemctl daemon-reload
  • 카프카 실행
    • systemctl start kafka-server.service
  • 상태 정보
    • systemctl status kafka-server.service
• 카프카 상태 확인
  • tcp 포트 확인
    • 주키퍼 netstat -ntlp | grep 2181
    • 카프카 netstat -ntlp | grep 9092
  • 주키퍼 지노드 이용한 정보 확인
    • /zookeeper경로/bin/zkCli.sh
    • 지노드 파일 확인 ls
  • 카프카 로그 확인
    • cat /kafka경로/logs/server.log
• 카프카 시작하기
  • 토픽생성
    • /kafka경로/bin/kafka-topics.sh --zookeeper peter-zk001:2181,peter-zk002:2181,peter-zk003:2181/peter-kafka --replication-factor 1 --partitions 1 --topic peter-topic --create
  • 삭제
    • /kafka경로/bin/kafka-topics.sh --zookeeper peter-zk001:2181,peter-zk002:2181,peter-zk003:2181/peter-kafka --topic peter-topic --delete
  • 프로듀싱
    • /kafka경로/bin/kafka-console-producer.sh --broker-list peter-kafka001:9092,peter-kafka002:9092,peter-kafka003:9092 --topic peter-topic
  • 컨슘
    • /kafka경로/bin/kafka-console-consumer.sh --bootstrap-server peter-kafka001:9092,peter-kafka002:9092,peter-kafka003:9092 --topic peter-topic --from-beginning