데이터 엔지니어링

카프카 구축 전에 예전에 정리했던 내용 기반으로 이론을 다시 정리해봤다. 

요즘에는 운영을 쉽게 하려고 MSK, 컨플루언트 카프카를 많이 사용 하는 것 같다.

아무튼 본론으로..

1.카프카란?

• 분산 이벤트 큐.
• 분산 이벤트 스트리밍 플랫폼
• 카프카 컨슈머, 프로듀서, 스트림즈, 커넥트 등 연동 API 제공
• 초당 수백만개 데이터를 처리할수 있으므로 빅데이터에 적합
• 분산 데이터를 통해 24시간 365일 안전하게 데이터를 처리할 수 있는 고가용성 기능 제공

2.왜 카프카?

• 고가용성
  • 서비스에 지장없는 운영을 보장.
• 낮은 지연
• 확장성
• 높은 처리량
  • 높은 처리량을 감당하지 못한다면, 서비스를 유지하기 힘듦

RabbitMQ , Redis 와의 차이점

• 이벤트 브로커
  • 서비스에서 발생한 이벤트를 브로커의 큐에 저장함
  • 딱 한번 일어난 이벤트 데이터를 브로커에 저장함으로써 단일 진실 공급원으로 사용 및 재처리가 가능
  • 마이크로 서비스 아키텍쳐에서 중요한 역할을 함
  • kafka , kinesis
• 메세지 브로커
  • 대규모 메세지 기반 미들웨어 아키텍쳐에서 사용
  • RabbitMQ, Redis

카프카 구조

• 기존 1:1 매칭으로 개발하고 운영하던 데이터 파이프라인은 커플링으로 인해 한쪽 이슈가 생기면 다른 한쪽에도 영향이 간다. → 카프카는 이러한 의존도를 타파하였다. (디커플링)
• 큐에 데이터를 보내는 것이 프로듀서이고 큐에서 데이터를 가져가는 것이 컨슈머다

카프카 특징

• 높은 처리량
  • 높은 처리량을 감당하지 못한다면, 서비스를 유지하기 힘듦
  • 우리 비지니스의 성공여부는 어떤 Threash hold 에 걸쳐지면 안된다.
  • 파티션 단위를 통해 동일 목적의 데이터를 여러 파티션에 분배하고, 이런 파티션을 컨슈머로 병렬처리할수 있는것이 큰 특징
  • 파티션 개슈만큼 컨슈머 개슈를 늘릴수 있다
• 확장성
• 영속성
  • 파일 io 성능 향상을 위해 os 에서 담당하는 페이지 캐시를 이용한다. 그래서 파일을 쓰고 읽는데도 빠를수 있다.
• 고가용성

3.카프카 구성 요소

토픽

• 구체화된 이벤트 스트림 = 쉽게 큐로 이해하면 됨
• 하나의 토픽에 여러 Producer / Consumer 가 존재할 수 있다.
• 토픽은 담는 데이터에 따라 이름을 줄 수 있다.

컨슈머

• 기본적으로 가장 오래된 순서대로 가져감 - 0번 오프셋부터
• 새로운 컨슈머가 구독을 하게 되도 가장 오래된 순서대로 가져감
  • auto.offset.reset = earliest 인경우

파티션

• 카프카의 토픽들은 여러 파티션으로 나눠짐.
• 파티션의 끝에서 0번 부터 차곡차곡 쌓이게 됨
• 토픽 = 논리적인 개념이라면, 파티션은 물리적인 저장소에 저장하는 단위
• 각 파티션은 Append-only 방식으로 기록됨
• 특정 파티션으로 데이터를 쓸수 있고, 명시되있지 않으면 RoundRobin 방식으로 파티션을 배정한다
• 파티션을 늘린다면?
  • 파티션을 다시 줄일수는 없다.
  • 컨슈머 개수가 늘어날때 분산 처리할 수 있다.
  • 신규 데이터는 2개의 파티션 중어디로 들어갈까?
  • 보통은 라운드로빈으로 파티션을 할당함
  • 키의 해시값으로
• 파티션 삭제 주기는?
  • log.retention.ms : 최대 record 보존 시간
  • log.retension.byte : 최대 record 보존 크기

오프셋

• 각각 파티션의 레코드는 Offset 식별자 정보를 가짐, 데이터 번호
• 카프카는 메세지 순서를 보장 하지 않음. 하지만 파티션이 1개라면 보장할지도?

4.카프카 클러스터

카프카 클러스터

• 카프카 브로커
  • 설치된 카프카의 서버 단위
  • 보통은 3대로 구성
• replication
  • replication 이 1인 상태라면 파티션이 브로커 서버에 1개만 저장된다.
  • replicaion 이 2라면 원본 하나, 복제본 1개의 파티션이 각각의 브로커 서버에 저장된다.
  • 따라서 replication 개수 ≤ 브로커 서버 개수
  • 원본 파티션 = Leader 파티션, 복제본 파티션 = follow 파티션
  • replication 의 설정된 값에 따라 서로 다른 브로커 서버에 파티션의 복제본이 생긴다.
• ISR(In-SyncReplica)
  • 리터 파티션의 레코드 개수 만큼 팔로워 파티션의 개수가 동일하게 복제가 된 안정된 상태
• ACK
  • 카프카 프로듀서는 ack 를 이용해 고 가용성 보장
    • ack = 0 , response 무시. 속도는 빠르지만 유실이 있음.
    • ack = 1, reponse 를 받음, 파티션 복제는 보장 못함. 유실 가능성 있음
    • ack = all , response 받고, follwer partition 저장확인 절차를 거침. 유실 가능성 없음
  • replication 개수가 늘어난다면 성능 저하.
• 파티셔너
  • 데이터를 토픽에 어떤 파티션에 넣는지 결정하는 역할을 함
  • 메세지 키 또는 메세지 값에 따라 파티션이 결정됨
  • hash(키) = 파티션 넘버
• 카프카 lag
  • 운영시에는 consumer lag이 발생
  • lag 이란 = 컨슈머가 마지막으로 읽은 offset - 프로듀서가 마지막으로 넣은 offset
  • 한개의 토픽과 컨슈머 그룹에 대한 lag 이 여러개 존재 하게 된다.
  • max lag 에 대한 모니터링이 운영시에는 필요하다
• lag burrow
  • golang 으로 개발된 오픈 소스
  • 컨슈머 lag 모니터링을 도와주는 독립적인 애플리케이션
  • 멀티 카프카 클러스터 지원
    • 2개이상의 카프카 클러스터를 운영할때, 하나의 burrow 로 운영 가능
• 주키퍼
  • 코디네이션 애플리케이션
  • 브로커 서버와 통신하며 상태관리, 컨슈머와의 통신, 카프카 메타데이터 정보를 저장함.

1.사전준비

• Docker , Docker Compose 가 설치되어 있어야 한다.
• 카프카 cli  사용을 위해 kafka binary 버전을 미리 받아 실행 준비를 해둔다.
  • https://kafka.apache.org/downloads

2. Docker-compose 작성

.env

CONFLUENT_VERSION=7.0.1

docker-compose.yml

version: '3'

networks:
  jssvs-net:
    external: true

services:
  zookeeper:
    image: confluentinc/cp-zookeeper:${CONFLUENT_VERSION}
    hostname: zookeeper
    restart: on-failure
    ports:
      - 2181:2181
    volumes:
      - ./zookeeper/data:/data
      - ./zookeeper/logs:/datalog
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000
    networks:
      - jssvs-net

  kafka-1:
    image: confluentinc/cp-kafka:${CONFLUENT_VERSION}
    hostname: kafka-1
    restart: on-failure
    ports:
      - 9091:9091
    depends_on:
      - zookeeper
    volumes:
      - ./kafka-1/data:/var/lib/kafka/data
    environment:
      KAFKA_BROKER_ID: 101
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_INTER_BROKER_LISTENER_NAME: "INTERNAL"
      KAFKA_ADVERTISED_LISTENERS: "INTERNAL://kafka-1:29092,EXTERNAL://localhost:9091"
      KAFKA_LISTENERS: "INTERNAL://:29092,EXTERNAL://:9091"
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: "INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT"
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 3
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 3
      CONFLUENT_METRICS_REPORTER_TOPIC_REPLICAS: 1
      CONFLUENT_METRICS_ENABLE: 'false'
    networks:
      - jssvs-net
  kafka-2:
    image: confluentinc/cp-kafka:${CONFLUENT_VERSION}
    hostname: kafka-2
    restart: on-failure
    ports:
      - 9092:9092
    depends_on:
      - zookeeper
    volumes:
      - ./kafka-2/data:/var/lib/kafka/data
    environment:
      KAFKA_BROKER_ID: 102
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_INTER_BROKER_LISTENER_NAME: "INTERNAL"
      KAFKA_ADVERTISED_LISTENERS: "INTERNAL://kafka-2:29092,EXTERNAL://localhost:9092"
      KAFKA_LISTENERS: "INTERNAL://:29092,EXTERNAL://:9092"
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: "INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT"
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 3
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 3
      CONFLUENT_METRICS_REPORTER_TOPIC_REPLICAS: 1
      CONFLUENT_METRICS_ENABLE: 'false'
    networks:
      - jssvs-net

  kafka-3:
    image: confluentinc/cp-kafka:${CONFLUENT_VERSION}
    hostname: kafka-3
    restart: on-failure
    ports:
      - 9093:9093
    depends_on:
      - zookeeper
    volumes:
      - ./kafka-3/data:/var/lib/kafka/data
    environment:
      KAFKA_BROKER_ID: 103
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_INTER_BROKER_LISTENER_NAME: "INTERNAL"
      KAFKA_ADVERTISED_LISTENERS: "INTERNAL://kafka-3:29092,EXTERNAL://localhost:9093"
      KAFKA_LISTENERS: "INTERNAL://:29092,EXTERNAL://:9093"
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: "INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT"
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 3
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 3
      CONFLUENT_METRICS_REPORTER_TOPIC_REPLICAS: 1
      CONFLUENT_METRICS_ENABLE: 'false'
    networks:
      - jssvs-net



  manager:
    image: obsidiandynamics/kafdrop
    restart: "no"
    ports:
      - 9000:9000
    environment:
      KAFKA_BROKERCONNECT: kafka-1:29092
      JVM_OPTS: "-Xms16M -Xmx48M -Xss180K -XX:-TieredCompilation -XX:+UseStringDeduplication -noverify"
    depends_on:
      - zookeeper
      - kafka-1
      - kafka-2
      - kafka-3
    networks:
      - jssvs-net

- 참조 (https://github.com/jaysooo/kafka-flink-stack-docker-compose)

3. 컨테이너 생성 및 실행

# 1) network 생성
$ docker network create jssvs-net

# 2) docker-compose up
$ docker-compose -f docker-compose.yml up -d

-> 브라우저에서 http://localhost:9000 로 접속하면 kafdrop 이라는 web UI 로 카프카 정보를 확인할 수 있다

4. 간단한 실습

카프카 binary 버전을 다운로드 받은후 하위  bin 디렉토리에 CLI 쉘 스크립트를 이용한다.

kafdrop 에서도 토픽 생성은 가능하다.

# 1) 토픽 생성
$ ./kafka-topics.sh --bootstrap-server 127.0.0.1:9091 --topic my-topic1 --create --partitions 1 --replication-factor 1

# 2) 토픽 정보 보기
$ ./kafka-topics.sh --bootstrap-server 127.0.0.1:9091 --topic my-topic1 --describe

# 3) 토픽 구독하기 (consumer)
$ ./kafka-console-consumer.sh --bootstrap-server 127.0.0.1:9091 --from-beginning --topic my-topic1

# 4) 토픽 삭제 하기
$ ./kafka-topics.sh --bootstrap-server 127.0.0.1:9091 --topic my-topic1 --delete

5. 마치며

• 회사에서 운영 목적으로 사용할 flink 를 공부하면서 카프카도 함께 공부 하고 있는데, 어렵기도 하고 시간이 너무 부족하다.
• 주로 실시간 데이터는 kinesis 로 받아서, firehose + lambda 기반 ETL 처리로 손쉽게 구성했었는데 아무래도 현업에서는 카프카가 더 많이 쓰이기도 하고, 최근에 트렌드가 데비지움 카프카 커넥트로 CDC -> kafka 구성을 많이 하는 것 같다.
• 다음 포스팅에서 카프카와 연동하는 producer application 과 consumer application 을 간단하게 작성해서 업로드 하겠다. 
• 바쁘다.. 바뻐  

1. 카프카란?

• 분산 이벤트 큐.
• 분산 이벤트 스트리밍 플랫폼
• 카프카 컨슈머, 프로듀서, 스트림즈, 커넥트 등 연동 API 제공
• 초당 수백만개 데이터를 처리할수 있으므로 빅데이터에 적합
• 분산 데이터를 통해 24시간 365일 안전하게 데이터를 처리할 수 있는 고가용성 기능 제공

2. 왜 카프카?

• 고가용성
  • 서비스에 지장없는 운영을 보장.
• 낮은 지연
• 확장성
• 높은 처리량
  • 높은 처리량을 감당하지 못한다면, 서비스를 유지하기 힘듦

3. 카프카 구성 요소

브로커

• 설치된 카프카의 서버 단위
• 보통은 3대로 구성

주키퍼

• 코디네이션 애플리케이션
• 브로커 서버와 통신하며 상태관리, 컨슈머와의 통신, 카프카 메타데이터 정보를 저장함.

토픽

• 구체화된 이벤트 스트림 = 쉽게 큐로 이해하면 됨
• 하나의 토픽에 여러 Producer / Consumer 가 존재할 수 있다.
• 토픽은 담는 데이터에 따라 이름을 줄 수 있다.

컨슈머

• 카프카와 통신하면서 메세지를 구독함
• 기본적으로 가장 오래된 순서대로 가져감 - 0번 오프셋부터
• 새로운 컨슈머가 구독을 하게 되도 가장 오래된 순서대로 가져감
  • auto.offset.reset = earliest 인경우

프로듀서

• 카프카와 통신하면서 메세지를 생산함

파티션

• 카프카의 토픽들은 여러 파티션으로 나눠짐.
• 파티션의 끝에서 0번 부터 차곡차곡 쌓이게 됨
• 토픽 = 논리적인 개념이라면, 파티션은 물리적인 저장소에 저장하는 단위
• 각 파티션은 Append-only 방식으로 기록됨
• 특정 파티션으로 데이터를 쓸수 있고, 명시되있지 않으면 RoundRobin 방식으로 파티션을 배정한다
• 파티션을 늘린다면?
  • 파티션을 다시 줄일수는 없다.
  • 컨슈머 개수가 늘어날때 분산 처리할 수 있다.
  • 신규 데이터는 2개의 파티션 중어디로 들어갈까?
  • 보통은 라운드로빈으로 파티션을 할당함
  • 키의 해시값으로 저장.
• 파티션 삭제 주기는?
  • log.retention.ms : 최대 record 보존 시간
  • log.retension.byte : 최대 record 보존 크기

오프셋

• 각각 파티션의 레코드는 Offset 식별자 정보를 가짐, 데이터 번호
• 카프카는 메세지 순서를 보장 하지 않음. 하지만 파티션이 1개라면 보장할지도?

파티셔너 

• 데이터를 토픽에 어떤 파티션에 넣는지 결정하는 역할을 함
• 메세지 키 또는 메세지 값에 따라 파티션이 결정됨
• hash(키) = 파티션 넘버

레플리케이션

• replication 이 1인 토픽은 하나의 브로커에만 저장됨.
• replicaion 이 2라면 원본 하나, 복제본 1개의 포티션이 2개의 브로커에 저장됨
• 따라서 replication 개수 ≤ 브로커 서버 개수
• 원본 파티션 = Leader 파티션, 복제본 파티션 = follow 파티션