$$config title: Kafka keywords: - Kafka $$ ## Kafka의 시작 - 처음은 단방향 통신에서 시작했지만 소스 애플리케이션과 타겟 애플리케이션이 많아져서 데이터를 조성하는 라인이 복잡 多 또한 데이터 전송시 프로토콜 파편화 심각 多 유지보수에 대한 어려움 多, Kafka 오픈소스 제공해서 사용함
## Kafka의 목적 - 카프카는 Source Application과 Target Application의 Coupling을 약하게 하기위해 만듦 ## 데이터의 흐름 - Source Application( EX) 쇼핑몰의 클릭로그, 은행 결제로그) 데이터 -> Apache Kafka에 데이터를 전송 -> Apache Kafka -> Target Application에 전송 Target Application EX) 로그적재, 로그처리등의 역할 Source Application에서 데이터의 종류는 제한 거의X EX) Json, tsv, Avro
## Kafka Topic이란 - Data를 최종 저장하는곳 (데이터를 구분하기 위한 장소) - DB Table과 File System과 유사한 성질을 갖고있다 - 여러개 생성 가능하다 - 큐(Queue) 와 비슷함 , queue란? 컴퓨터의 기본적인 자료 구조의 한가지로, 먼저 집어 넣은 데이터가 먼저나오는 구조로 저장하는 형식 - item을 리스트 끝부분에 추가 / 제거는 첫 번째 항목부터 ->ex) 추가할때는 5 4 3 2 1 / 삭제할때는 ? 4 3 2 1 - Producer와 Consumer는 Library로 구성되어있어서 Application에서 구현 가능 ## Kafka의 장점 - Data 흐름에서 fault tolerant(고가용성) ->시스템내의 어느 한 부품 또는 어느 한 모듈에 Fault (장애)가 발생하더라도 시스템운영에 전혀 지장을 주지 않도록 설계된 컴퓨터 시스템 - Server Issue or Power Isuue 에도 **Data 손실 X, 복구 가능, 낮은 지연, 높은 처리량, 데이터를 많이 처리할수있음** 그렇기 때문에 빅데이터를 사용하는곳에서는 Kafka를 안쓰는곳이 없음
## Topic의 특징과 전달 과정 - Producer가 Topic에 데이터 넣음 -> Consumer가 Topic에서 데이터 가져옴 - Topic은 이름을 가질 수 있어서 역할(목적)에 따라 정확하게 무슨 데이터를 담는지 명확하게 지으면 유지보수 시에 편리하게 관리 할 수 있다. - 하나의 Topic은 여러개의 파티션 구성이 가능하다. - 1개 파티션은 Queue와 같이 파티션 끝 -> 0000011 이와같이 쌓이게된다. - Consumer가 붙으면 가장 오래된 것 부터 순서대로 가져간다. - **Consumer가 데이터를 가져가게되도 파티션에 데이터는 그대로 남는다** - **새로운 Consumer가 붙었을 때 남아있는 Pation에서 다시 데이터를 가져올 수 있다.** 그러나 Consumer의 group이 달라야 하고 세팅값이 auto.offset.reset=earliest 이여야 한다.
- **위와 같이 동일 데이터를 여러번 처리 가능하다는 것이 중요하다.** - 파티션의 개수가 여러개면 데이터를 넣을때 키 지정이 가능하다 1. key가 null, 기본 파티션을 사용할 경우 라운드로빈으로 할당한다. @라운드 로빈은 그룹 내에 있는 모든 요소들을 합리적인 순서에 입각하여 뽑는 방법으로서, 대개 리스트의 맨 위에서 아래로 가며 하나 씩 뽑고, 끝나면 다시 맨 위로 돌아가는 식으로 진행된다. 쉽게 말해 라운드 로빈은 "기회를 차례대로 받기"라고 이해해도 좋을 것이다. 2. key가 null이 아니고, 기본 파티셔너를 사용할 경우 Key의 해시(Hash)값을 구해 특정 파티션에 할당한다 ## 파티션을 늘리는 이유와 주의사항 - 파티션을 늘리는 것은 가능하지만 줄이는것은 불가능하기때문에 주의해야한다 - **파티션을 늘리는 이유**는 파티션 개수를 늘리고 컨슈머의 개수를 늘려 **데이터 처리를 분산** 시키기 위함이다
- 그럼 파티션의 Record 삭제는 언제하나? option에 따라 다르지만 Record 저장되는 최대시간과 크기가 지정 가능하다. 지정하게되면 일정기간 데이터 저장이 가능하고, 적절하게 데이터가 삭제되도록 설정이 가능하다
## Kafka Producer - Data를 Kafka에 대량으로 보내는 역할을 한다 EX) 방대한 양의 Click Log - Data를 producing, 즉 Data를 생산하는 역할이다. - Topic에 전달할 Data 생성. - 특정 Topic으로 데이터 전송 가능. - **Kafka broker로 Data 전송시 성공 / 실패 여부를 알 수 있고 실패시에 재시도가 가능하다.** - Kafka Consumer와 Producer를 사용하기 위해서는 Library추가가 필요하다. 메이븐(Maven)과 그래들(Gradle)같은 도구를 사용하여 가져올 수 있다. ## dependency로 가져올때 주의사항 - **Version 주의** - broker 버전과 클라이언트 버전의 하위 호환성이 완벽하게 모든버전을 지원하지 **않는다.** 일부 Kafka broker버전은 특정 Kafka 버전을 지원 안 할 수도 있다 - **브로커와 클라이언트 버전의 하위 호환성에 대해 숙지하고 알맞는 Kafka를 사용해야 한다. ## JAVA에서 Kafka Setting - java application부터 Spring boot까지 kafka-client library를 추가하여 사용할 수 있음 - **Producer 세팅** ```js public class Producer{ public static void main(String[] args) throws IOException{ // properties config 세팅 시작 Properties configs = new Properties(); configs.put("bootstrap.servers", "localhost:9092"); //부트스트랩 서버 설정 (localhost의 카프카를 바라보도록 설정) configs.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); configs.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer"); //properties config 세팅 끝 KafkaProducer producer = new KafkaProducer (configs); // 카프카 프로듀서 인스턴스 생성 ProducerRecord record = new ProducerRecord ("click_log", "login"); //카프카 클라이언트에서 ProducerRecord 클래스를 제공해준다. click_log토픽에 login이라는 value를 보낸다고만 설정함 //ProducerRecord 인스턴스 생성시 어느토픽에 넣을것인지, 어떤 key와 value를 담을것인지 선언 할 수 있다. 자세한것은 하단에 api공식문서 참고 // 위와 다르게 key를 포함하여 보내고 싶다면 new ProducerRecord ("click_log", "1", "login"); 이와같이 선언하면됨 파라미터 개수에 따라 알아서 오버로딩되어 사용하기때문에 의도와 맞게 api문서를 참고하여 사용 producer.send(record); //send()메소드의 파라미터로 ProducerRecord의 새로운 인스턴스를 넣으면 전송된다. producer.close(); //전송이 완료되면 프로듀서를 종료해준다 } } ``` - [ProducerRecord API](https://kafka.apache.org/23/javadoc/org/apache/kafka/clients/producer/ProducerRecord.html), [KafkaProducer API](https://kafka.apache.org/23/javadoc/index.html?org/apache/kafka/clients/producer/KafkaProducer.html) - 첫 번째 세팅을 보면 bootstrap.server 설정을 볼 수 있는데 2개 이상의 IP와 Port 설정을 권장한다 **둘중 하나가 비정상일 경우 다른 정상적인 브로커에 연결되어 사용 가능하기 때문이다 실제 애플리케이션과 연동해 사용할때는 위와같이 사용하지말고 반드시 두개 이상의 브로커 정보를 넣는것을 추천한다.** - 두 세번째 세팅에는 Key와 Value가 들어가는데 StringSerializer를 사용해 직렬화를 설정한다 Serializer란? Key 혹은 value를 직렬화하기 위해 사용되고 Byte array, String, Integer 시리얼라이즈를 사용할수 있다. - key는 메세지를 보내면, 토픽의 파티션이 지정될 때 쓰인다.
- key값을 주지않은 데이터를 파티션 한개에 보내게되면 우측 끝에서부터 차곡차곡 쌓이게된다. 하지만 파티션이 한개가 아니라 두개라면? 라운드로빈으로 2개의 파티션에 차곡 차곡 쌓이게된다. 한쪽의 파티션에 데이터가 있을때 양쪽의 개수를 맞추기보다는 그냥 차례대로 0번지파티션 1번지파티션에 하나씩 쌓이게 된다.
- 만약 record에 키 값이 있을때는 카프가가 key를 특정한 hash값으로 변경시켜 파티션과 1 대 1 매칭을 시켜준다. 그렇기때문에 2개의 프로듀서와 2개의 파티션이 있을땐 각각의 키값에 따라 구분되어 각각의 파티션에 쌓이게된다. 이 때 토픽에 파티션을 하나 더 추가하게되면 key와 파티션의 1대1 매칭이 깨지기 때문에 key와 파티션의 매칭이 보장되지않으니 추후에도 생성하지 않는것을 권장한다.
- **위와같이 카프카 프로듀서의 기본적인 구현은 간단하지만 데이터 유실 혹은 브로커의 이슈에 대처하기 위해서는 추가적인 옵션들과 코드가 필요하다.**
## Kafka의 중요 요소 - **broker, replication, ISR**
- broker는 카프카가 설치되어 있는 서버 단위를 말한다. 보통 3개 이상의 broker로 구성하여 사용하는 것을 권장한다. - **replication(복제)** 는 카프카의 아키텍쳐의 핵심이다. 클러스터에서 서버에 장애가 생길 때 카프카의 가용성을 보장하는 가장 좋은 방법이 복제이기 때문이다.
- partition 1개, replication 1개 = 브로커(서버 단위) 한개에 원본만 유지 / replication의 개수가 많아지면 다른 브로커에 그만큼 복제가 많아지는데 브로커 개수보다 replication 개수가 더 많아질 수는 없다. 이 때 ISR이라는 개념을 쓴다 ISR이란 In Sync Replication의 줄임말인데 브로커가 총 3개에 replcation이 3개라면 원본 1개 partition을 Reader partition이라고 부른다. 그리고 나머지 2개를 Follower partition이라고부른다. 이 Leader, Follower partition을 합쳐서 ISR(InSyncReplication) 이라고 부른다.
- 왜 replication을 사용하는건가? partion의 고가용성을 위해서다. 만약 브로커가 3개고 replication이 1, partition이 1인 topic이 존재한다고 하면 브로커가 사용이 불가해질때 그 파티션은 복구 할 수가 없다. 하지만? raplication이 2 이상이라면 Follower partition이 존재하므로 데이터 복구가 가능해진다 이 때 Follower partition이 Leader partion 으로 역할을 승계하게 된다.
- 프로듀서에는 ack라는 상세 옵션이 있는데 이를 통해 고가용성을 유지할 수 있다 ack는 0, 1, all 옵션 3개중 선택 할 수 있으며 0일때는 Reader partition에 데이터를 전송하고 응답값을 받지 않는다. 그렇기 때문에 Leader partition에 데이터가 정상적으로 전송됐는지, 나머지 partition에 복제가 잘 되었는지 알 수 없다. 이 때문에 속도는 빠를 지 모르지만 데이터 유실 가능성이 있다.
- 1일경우 리더 파티션이 데이터를 정상적으로 받았는지 응답값은 받지만 나머지 파티션에 복제가 되었는지는 알 수 없다. 그래서 만약 리더 파티션이 데이터를 받는 즉시 브로커가 고장난다면 나머지 파티션에 데이터가 전송되지 못한 상태이므로 0 옵션과 마찬가지로 데이터 유실 가능성이 있다.
- all일경우 모든 follwer 파티션에게 응답을 받는다. 그렇기 때문에 데이터 유실 가능성은 없다 / 하지만 확인하는 부분이 많기 때문에 속도가 현저히 느리다는 단점이 있다.
- 그럼 replikcation이 많으면 많을수록 좋은거 아닌가 라고 생각할 수 있지만 개수가 많아지면 그만큼 브로커의 리소스 사용량이 늘어나게 되기 때문에 카프카에 들어오는 데이터 량과 저장시간을 잘 고려해서 replication 개수를 정하는게 좋다 - 3개 이상의 브로커를 사용할때 replication은 3으로 설정하는것을 권장한다. # acks

옵션	메시지 손실률	속도	설명
acks = 0	상	상	프로듀서는 자신이 보낸 메시지에 대해 카프카로부터 확인을 기다리지 않습니다.
acks = 1	중	중	프로듀서는 자신이 보낸 메시지에 대해 카프카의 leader가 메시지를 받았는지 기다립니다. follower들은 확인하지 않습니다. leader가 확인응답을 보내고, follower에게 복제가 되기 전에 leader가 fail되면, 해당 메시지는 손실될 수 있습니다.
acks = all(-1)	하	하	프로듀서는 자신이 보낸 메시지에 대해 카프카의 leader와 follower까지 받았는지 기다립니다. 최소 하나의 복제본까지 처리된 것을 확인하므로 메시지가 손실될 확률은 거의 없습니다.

# Kafka 컨슈머 - Kafka 컨슈머의 동작은 다른 메시징 시스템들과는 다소 차이가 있는데 다른 메시징 시스템에서는 컨슈머가 데이터를 가져가면 큐 내부 데이터가 사라지는데 카프카에서는 컨슈머가 데이터를 가져가더라도 **데이터가 사라지지 않는다.** 이와 같은 특징은 카프카 그리고 카프카 컨슈머를 데이터 파이프라인으로 운영하는데 매우 핵심적인 역할을 한다.
- 카프카 컨슈머는 기본적으로 토픽 내부의 파티션에서 저장된 데이터를 가져온다. 이를 폴링(polling)이라고 한다.
- **Counsumer의 역할** - Topic의 partition으로 부터 데이터 polling / 메시지를 가져와서 특정 데이터베이스에 저장 or 또 다른 파이프라인에 전달 - Partition의 offset 위치 기록(commit) / offset이란 파티션에 있는 데이터의 번호(순서)를 뜻하는데, 만약 컨슈머가 실행이 중지되었다고하면 나중 실행시 오프셋을 읽어 어디까지 읽었나 consumer_offset을통해 데이터처리 시점 복구 가능 - Consumer group을 통해 병렬처리 / 컨슈머가 여러개일 경우 병렬처리 가능 -> 파티션 개수에 따라 컨슈머를 여러개 만들면 병렬처리가 가능하기 때문에 더욱 빠른속도로 데이터를 처리 할 수 있다. - ```js //gradle 기준 compile group: 'org.apache.kafka', name: 'kafka-clients', version: '2.3.0' //maven 기준 org.apache.kafka kafka-clients 2.3.0 ``` - 카프카는 브로커 버전과 클라이언트 버전의 호환성을 확인 후 사용 [카프카 버전별 하위호환설명 링크](https://blog.voidmainvoid.net/193)
- Consumer properties 설정 ```js public Class Consumer{ public static void main(String[] args){ //세팅 시작 Properties configs = new Properties(); configs.put("bootstrap.servers", "localhost:9092" ); // 카프카 브로커 설정 -> 1개만 적어놨지만 여러개여야 이슈가 생겼을때 다른 브로커에 붙어 사용할 수 있음 configs.put(group.id", "click_log_group"); //그룹아이디 ==컨슈머 그룹이라고도 부르는데 컨슈머 그룹은 컨슈머들의 묶음이라고 생각하면됨 configs.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); // Producer 세팅과 같이 key, value에 대한 직렬화 configs.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); //세팅 끝 KafkaConsumer consumer = new KafkaConsumer(configs); //프로듀서 인스턴스 생성과 거의 일치하게 컨슈머 생성 consumer.subscribe(Arrays.asList("click_log")); // 상단에서 어떤 카프카 브로커에서 가져올지 선언했으니 여기서는 어느 토픽을 대상으로 할지 정함 // 만약 특정 토픽의 전체 파티션이 아니라 일부 파티션의 데이터만 가져오고싶다면 하단과 같이 assign()메소드를 사용하면된다. // 예시시작 TopicPartition partition0 = new TopicPartition(topicName, 0); //만약 key가 존재하는 데이터라면 이 방식을 통해 데이터의 순서를 보장하는 데이터 처리가 가능하다. TopicPartition partition1 = new TopicPartition(topicName, 1); consumer.assign(Arrays.asList(partition0, partition1) ); // 예시끝 while(ture){ //폴링루프 구문( poll() 메소드가 포함된 무한루프) == 컨슈머 API의 핵심 로직 -> 브로커로부터 연속적으로 그리고 컨슈머가 허락하는 한 많은 데이터를 읽는 것 ConsumerRecords records = consumer.poll(500); // 밀리세컨드 0.5초 설정한 시간동안 데이터를 기다렸다 코드 실행함 // 0.5초동안 데이터가 안들어오면 빈값의 records 변수 반환하고 데이터가 있으면 데이터가 존재하는 records 값을 반환한다. // records 변수는 데이터 배치로서 레코드의 묶음 list이다 그러므로 실제로 카프카에서 데이터를 처리할 때는 가장 작은 단위인 record로 나누어 처리하도록한다 for(ConsumerRecord record : records){ System.out.println(record.value()); // <- 실제기업에서는 하둡이나 엘라스틱서치와 같은 저장소에 저장하는 로직이 들어갈것임 } } } } ``` - 파티션이 2개인데 컨슈머가 1개인경우에는 2개의 파티션에서 데이터를 가져간다 / 2개의파티션 2개의컨슈머일때는 컨슈머가 각각의 파티션을 할당하여 가져간다. 만약 2개파티션 3개 컨슈머일때는 이미 파티션들이 각 컨슈머에 할당되었기 때문에 더이상 할당될 파티션이 없어서 컨슈머 한개는 동작하지않는다 **그러므로 컨슈머의 개수는 파티션 개수보다 적거나 같아야한다**
- 예시로 데이터 실시간 시각화 및 분석을 위해 Elasticsearch에 데이터를 저장하는 역할을 하는 컨슈머 그룹이 있고 추가로 데이터 백업 용도로 hadoop에 데이터를 저장하는 컨슈머 그룹이 새로 들어와도 컨슈머 그룹이 다르기 때문에 각 파티션에 특정 offset의 데이터를 읽고있는것들은 서로 영향이 없다. 왜냐하면 _consumer_offset 토픽에는 컨슈머 그룹별로 토픽별로 offset을 나누어 저장하기 때문이다. **이러한 카프카의 특징을 토대로 하나의 토픽으로 들어온 데이터는 다양한 역할을 하는 여러 컨슈머들이 각자 원하는 데이터로 처리가 될 수 있다.**