WantAirpod

카프카란?

분산 이벤트 스트리밍 플랫폼이다.

Before Kafka

처음에는 이 처럼 단반향(source->target) application으로 시작했다.

하지만 점차 요구사항이 많아지면서 파편화가 심해졌다.

다음과 같이 매우 복잡한 구조로 변모했다. 추후에 변경이 발생하였을 때 유지 보수하기 매우 어려워 졌다.

이러한 어려운 점을 해결하기 위해 링크드인에서 내부적으로 개발하였고 현재는 오픈소스로 뿌려줬다.

카프카의 탄생배경

2011년 링크드인에 의해 처음 공개 되었다. 링크드인은 당시 데이터파이프라인 처리를 위해 기존 redis나 rabbitMQ등 여러 애플리케이션들을 혼용해서 사용해야 했다. 하지만 이런 애플리케이션 각각의 특징이 워낙 뚜렷하다보니 데이터 처리에 있어 파편화가 심각했다.

카프카는 데이터 처리를 각각 여러 애플리케이션에서 처리하는 것이 아니라 중앙 집중화 및 실시간 스트림 처리한다. (=중추신경)

실시간 데이터 처리와 데이터 파이프라인으로 용이한 시스템

왜 많이 사용할까?

• High throughput message capacity
  • 짧은 시간 내에 엄청난 양의 데이터를 컨슈머까지 전달 가능하다.
  • 파티션을 통한 분산 처리가 가능하기 때문에 데이터 양이 많아 질수록 컨슈머 갯수를 늘려 병렬처리가 가능하고 이를 통해 데이터 처리를 더욱 빠르게 할 수 있습니다. 
• Scalability와 Fault tolerant
  • 확장성이 뛰어나다. 이미 기존에 카프카 브로커가 있다해도 브로커를 추가해서 수평 확장이 가능하다. 
  • 이렇게 수평확장된 브로커가 죽더라도 이미 replica로 복제된 데이터는 안전하게 보관되어 있으므로 복구하여 처리가능하다.
• Undeleted log
  • 다른 플랫폼과 달리 카프카 토픽에 들어간 데이터는 컨슈머가 데이터를 가지고 가더라도 데이터가 사라지지 않습니다. 
  • 다른 플랫폼에서는 복잡한 방식이다.
  • 하지만 카프카에서는 컨슈머의 그룹 아이디만 다르다면 카프카 토픽에 들어간 데이터를 손실하지 않고 재사용이 가능하다.

데이터 처리가 많은 카카오, 네이버에서만 사용 가능할까?

아니요. 카프카는 소규모 스타트업에서 적용하기에도 굉장히 좋다. 이유는 스타텁에서 중요한 것은 안정적인 운영과 빠른 확장성이다. 소규모로 시작하여도 추후를 생각해 확장성이 용이하다는 점이 좋다.( = scale out : 카프카 클러스터 내부의 브로커 갯수를 늘려서 원활하게 데이터를 처리할 수 있습니다.)

Kafka features

카프카는 source와 target의 커플링을 약하게 하기 위해서 탄생했다.

source단에서 클릭로그, 결제로그를 다양한 포맷으로 전달할 수 있다.

카프카는 다양한 Topic 을 담을 수 있는데 쉽게 생각해서 queue라고 볼 수 있다.

queue에 데이터를 넣는 역할은 Producer가 하고 queue에서 데이터를 가져가는 역할은 Consumer가 한다. 둘다 라이브러로 되어있으므로 언제든지 구현 가능하다. 결론적으로 Kafka는 아주 유연한 고가용성 queue라고 볼 수 있다. 

Kafka Topic

카프카에서는 토픽을 여러개 생성할 수 있다. 토픽은 데이터 베이스의 테이블이나 파일시스템의 폴더와 유사한 성징을 가지고 있다.

Kafka의 토픽은 다음과 같이 click_log, send_log 등 이름을 명시하면 유지 보수시 편리하게 관리할 수 있다.

Kafka Topic의 내부

하나의 토픽은 여러개의 파티션으로 구성될 수 있으며 첫 번째 파티션 번호는 0번부터 시작한다.

컨슈머는 가장 오래된 순서대로 가져간다.(선입선출) 더 이상 데이터가 없다면 또 다른 데이터가 들어 올때까지 대기한다. 

특이한 점은 컨슈머가 데이터를 가져가더라도 데이터는 삭제 되지 않고 그대로 남는다.

그렇다면 남는 데이터들은 누가 가져가는 것일까? 바로 새로운 컨슈머가 붙었을 때 다시 0번부터 가져가서 사용가능하다.

(단, 다른 컨슈머여야 하고 auto.offset.reset = earliest 여야 합니다.)

이처럼 사용할 경우 동일한 데이터에 대해서 2번 사용이 가능한데 이는 카프카를 사용하는 아주 중요한 이유이기도 하다.

* 클릭로그를 분석하기 위해 ES에 저장하기도 하고, 클릭로그를 백업하기 위해 Haddop에 저장하기도 한다. 각각은 다른 기능을 위해서 사용하는 topic이다.

만약 다음과 같이 파티션을 하나 더 늘렸다면(7번 직전에) 7번은 어디로 들어갈까?

1) 만약 키가 null이고, 기본 파티셔너 사용할 경우 -> 라운드 로빈

2) 만약 키가 있고, 기본 파티셔너 사용할 경우

-> 키의 해시(hash) 값을 구하고, 특정 파티션에 할당 

따라서 해당 경우는 라운드 로빈에 의해서 파티션#1에 할당된다.

그 이후에 다음과 같이 라운드 로빈으로 쭉 적재된다.

파티션을 늘리는 것은 아주 조심해야한다.

• 이유는 파티션은 늘리는 것은 마음대로이지만 줄이는 것은 불가능하다. 왜 그러면 파티션을 늘리는 것일까? 파티션을 늘리면 컨슈머의 갯수를 늘려서 데이터 처리를 분산 시킬 수 있다.

이렇게 데이터가 늘어나면 파티션의 데이터는 언제 삭제 될까요?

• 삭제되는 타이밍은 옵션에 따라 다르다. 다음을 저장하면 일정한 기간, 용량 동안만 데이터를 저장하고 적절하게 삭제 될 수있다.
  • log.retentios.ms : 최대 record 보존 시간
  • log.retention.byte : 최대 record 보존 크기(byte)

카프카 프로듀셔

프로듀서는 데이터를 카프카에 보내는 역할을 한다. 예를들어 엄청난 양의 클릭로그들을 대량으로, 그리고 실시간으로 카프카에 적재할 때 프로듀셔를 사용할 수 있다.  

프로듀서는 이름 그대로 데이터를 생산하는 역할을 합니다. 즉 데이터를 kafka topic에 생성한다는 말이다. 

카프카 프로듀서의 역할

• Topic에 해당 메시지를 생성
• 특정 Topic으로 데이터를 publish
• 처리 실패 / 재시도

카프카를 사용하기 위해서는 위와 같이 아파치 카프카 라이브러리를 추가 해야한다. 주의해야할 점은 버전입니다. 카프카는 브로커 버전과 클라이언트 버전의 하위호환성이 완벽하게 모든 버전에 대해서 지원하지는 않습니다.

하위 호환성 참고 : https://blog.voidmainvoid.net/193

카프카 구현

자바 프로퍼티 객체를 통해 프로듀서의 설정 정의

//자바 프로퍼티 객체를 통해 프로듀서의 설정 정의
Properties configs = new Properties();
configs.put("bootstrap.servers", "localhost:9092");
configs.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
configs.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

• 부트 스트랩 서버 설정을 localhost:9092를 바라보게 설정
  • 카프카의 주소 목록(ip, port)을 되도록 2개 이상 권장 --> 둘 중 하나의 브로커가 고장나도 다른 한쪽으로 로드밸런싱 하기 위해서이다. 
• Key와 Value 설정을 StringSerializer로 직렬화하여 삽입한다.
  • 시리얼라이저는 키 혹은 value를 직렬화하기 위해 사용하는데 Byte array, String, Integer 시리얼라이즈를 사용할 수 있습니다. 
  • 키는 메시지를 보내면, 토픽의 파티션이 지정될 때 쓰인다. 

카프카 프로듀서 인스턴스 생성

//카프카 프로듀서 인스턴스 생성
KafkaProducer<String, String> producer = new KafkaProducer<String, String>(configs);

전송할 객체 생성 및 전송

//전송할 객체 생성 및 전송
ProducerRecord record = new ProducerRecord<String,String>("click_log" , "login");
//키 포함
ProducerRecord record = new ProducerRecord<String,String>("click_log" ,"1","login");
producer.send(record);
producer.close();

• click_log 토픽에 login 이라는 value를 보낸다. 만약 키를 포함하고 싶을 경우 2번째 파라미터에 담아서 보낸다. (위에서는 1)
• send 메서드 : send 메서드에 Recod를 보내면 전송이 이루어진다.
• close 메서드 : 종료해준다.

	Input(프로듀서)	파티션 내부	설명
경우 1 : key null 인 경우			- key가 null인 경우, 파티션#0에 보내는 경우 - 아래와 같이 한개 씩 쌓이기 시작한다. - 만약 파티션이 2개이면 라운드 로빈 방식으로 차곡 차곡 쌓인다.
경우 2 : key 있는 경우			-key가 각각 1, 2 이다. 카프카는 key를 특정한 hash값으로 변경 시켜 파티션과 1:1 매칭을 시킵니다.  - 만약 파티션이 한개 더 추가되면 어떻게 될까? 키와 파티션의 갯수가 일치하지 않기 때문에 매칭이 보장되지 않고 일관성 없게 쌓인다.

카프카 브로커와 Replication, ISR

카프카의 핵심 구성요소이다.

카프카가 설치되어 있는 서버 단위를 말한다. 보통 3개 이상의 브로커로 구성하여 사용하는 것을 권장한다. 

만약 파티션이 1개이고 replicatiion이 1인 topic이 존재하고 브로커가 3대라면 브로커 3대 중 1대에 해당 토픽의 정보(데이터)가 저장됩니다.

replication(복제)

replication은 partition의 복제를 뜻합니다. 

replication이 1인 경우

replication이 1이라면 partition(복제본)이 한개만 존재한다는 뜻이다. 

Replication이 2이 인 경우

만약 replication이 2라면 partition은 원본 1개와 복제본 1개로 총 2개가 존재한다.

Replication이 3이 인 경우

만약 replication이 2라면 partition은 원본 1개와 복제본 2개로 총 3개가 존재한다. 원본 1은 Leader Partition, 복제본 2개는 Follower partition이다. 

단, broker에 갯수에 따라서 replica의 갯수가 제한되는데 브로커갯수가 3이면 replica은 4가 될수 없다는 뜻이다. 즉 replication <= broker 이다.

ISR(In Sync Replica)

leader partition + follower partition = ISR

왜 Replication을 사용할까?

Replication은 partition의 고가용성을 위해 사용한다. replication 2 이상일 경우는 브로커 한개가 죽더라도 다른 복제본 브로커로 돌릴수 있기 때문에 사용하면 좋다. follwer partition이 leader partition역할을 승계하게 되는 것 입니다. 

프로듀서의 Ack 옵션

프로듀서가 토픽의 파티션에 데이터를 전달할 때, 전달받는 주체가 바로 Leader Partition이다.

프로듀서는 ack = 0, 1, all 세가지 상세 옵션이 있다. 

Ack = 0 인 경우

Leader Partition에 값을 전달한 후 응답 값을 받지 않습니다. 따라서 데이터 유실이 일어났는지 알 수없습니다. 단 속도는 빠릅니다.

Ack = 1 인 경우

1인 경우는 Leader partition에 데이터를 전송하고 응답은 받지만 적절하게 replication 되었는지는 확인하지 않아 데이터 유실가능성이 있습니다. 단 속도는 빠릅니다. 

Ack = 2 인 경우

replication까지 복제가 잘 되었는지 확인하는 절차까지 받습니다. 따라서 데이터 유실가능성은 없다. 단 속도가 현저히 느리다.

Replication 갯수가 많아지면..?

replication이 고가용성을 위해 중요한 역할을 한다면 broker갯수 만큼 무조건 늘리는게 답일까? 

아니다. 그렇다면 브로커의 리소스 사용량도 늘어나게 된다. 즉, 적당한게 좋다. 3개이상의 브로커를 사용하는 경우 replication을 3으로 해두는 것을 추천한다.

카프카 설치 및 실행

brew를 이용하여 kafka를 다운로드한다. 카프카를 사용하기 위해 다양한 디펜던시가 있고 그중에 대표적인 것이 zookeeper이다.

brew install kafka

카프카를 실행하기 위해서는 먼저 zookeeper를 실행한다.

brew services start zookeeper

카프카를 실행한다.

brew services start kafka

토픽을 만들고 데이터를 넣고, 조회하기

먼저, 카프카에서는 다양한 script를 제공한다. 하단의 명령을 통해 script를 확인 할 수 있다. 하단의 명령어를 통해 카프카가 깔린 곳으로 이동한다.

brew info kafka
.../usr/local/Cellar/kafka/3.1.0
cd /usr/local/Cellar/kafka/3.1.0/bin

토픽 생성 명령어 (파티션 1, replication 1 인 토픽)

./kafka-topics --create --zookeeper localhost:2181 
--replication-factor 1 -partitions 1 --topic cjy9249

local에 띄어진 카프카로 string 데이터를 전송할 수 있다.

./kafka-console-producer --broker-list localhost:9092 --topic cjy9249
hello //입력
kafka //입력

토픽에 들어가 있는 데이터 읽어오기

./kafka-console-consumer --bootstrap-server localhost:9092 --topic cjy9249 --from-beginning
hello //출력
kafka //출력

참고

라인 개발자의 카프카 활용

https://engineering.linecorp.com/ko/blog/how-to-use-kafka-in-line-1/

메시지 브로커

• 이벤트 브로커로 역할을 할 수 없다.
• 많은 기업들의 대규모 메시지 기반 미들웨어 아키텍처에서 사용

ex) 메시징 플랫폼, 인증 플랫폼, 데이터베이스

*미들웨어
서비스하는 애플리케이션들을 보다 효율적으로 아키텍처들을 역결하는 요소들로 작동하는 소프트웨어를 뜻한다.

• 메시지 브로커에 있는 큐에 데이터를 보내고 받는 프로듀서와 컨슈머를 통해 메시지를 통신하고 네트워크를 맺는 용도로 사용해왔습니다.

특징

• 메시지를 받아서 적절히 처리하고 나면 즉시 또는 짧은 시간 내에 삭제되는 구조이다.
• 데이터를 보내고, 처리하고 삭제한다.

ex ) 레디스 큐, 레빗엠 큐

이벤트 브로커

• 이벤트 브로커는 메시지 브로커 역할을 할 수 있다.

특징

• 이벤트 또는 메시지라고도 불리는 이 레코드 이 장부를 딱 하나만 보관하고 인덱스를 통해 개별 엑세스를 관리합니다. 
• 업무상 필요한 시간동안 이벤트를 보존할 수 있다. 
• 이벤트 브로커는 삭제하지 않는다. 

왜 데이터를 처리했는데 삭제하지 않을까?
단서는 '이벤트' 라는 단어에 숨어있다. 이벤트 브로커는 서비스에서 나오는 이벤트를 마치 데이터 베이스에 저장하듯이 이벤트 브로커의 큐에 저장하는데요. 이렇게 저장하면서 얻는 명확한 이점이 있다. 
첫번째 이점은 딱 한번 일어난 이벤트 데이터를 브로커에 저장함으로써 단일 진실 공급원으로 사용할 수 있다.
두번째는 장애가 발생했을 때 장애가 일어난 시점부터 재처리할 수 있습니다. 
세번째는 많은 양의 실시간 스트림 데이터를 효과적으로 처리할 수 있다는 특징이 있다. 
그외에도 다양한 MSA에서 중요한 역할을 맡을 수 있다.  

ex) 카프카나 aws 키네시스가 대표적이다. 

이벤트 브로커로 클러스터를 구축하면 이벤트 기반 마이크로 서비스 아키텍처로 발전하는데 아주 중요한 역할을 한다. 메시지 브로커로서도 사용할 수 있으니깐 팔방미인이다. 

카카오에서의 도입

도입 전문