# Begin Kafka

## Begin Kafka

[아파치 카프카 for beginners](https://www.inflearn.com/course/%EC%95%84%ED%8C%8C%EC%B9%98-%EC%B9%B4%ED%94%84%EC%B9%B4-%EC%9E%85%EB%AC%B8/dashboard) 강의를 요약한 내용입니다.

[https://kafka.apache.org/](https://kafka.apache.org/)

## Kafka?

**Before Kafka**

> 소스 애플리케이션, 타겟 애플리케이션 개수가 늘어날 수록 데이터 라인이 많아지게 된다.

* 이 때 배포와 장애 대응이 어려워지고,
* 데이터 전송 시 프로토콜 포맷 파편화가 심해짐
* 추후 데이터 포맷에 변경사항이 있을 경우 유지보수에 어려워진다.

<figure><img src="https://3925856803-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2FfJN93IwN0ptPSTmZ53eY%2Fuploads%2Fgit-blob-0ebc4df29b974fe9261448e53d9a0a951f069393%2Fbefore-kafka.png?alt=media" alt=""><figcaption></figcaption></figure>

After Kafka

> 소스/타겟 애플리케이션의 커플링을 약하게 하기 위해 출시

* 위와 같은 복잡함을 해결하기 위해 링크드인에서 내부적으로 개발하여 오픈소스로 제공

<figure><img src="https://3925856803-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2FfJN93IwN0ptPSTmZ53eY%2Fuploads%2Fgit-blob-472aaa96219c4c424dfedd48cc4581437358820c%2Fkafka.png?alt=media" alt=""><figcaption></figcaption></figure>

<figure><img src="https://3925856803-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2FfJN93IwN0ptPSTmZ53eY%2Fuploads%2Fgit-blob-c4dbcac6a2890c027d7cc782488131254251dfb7%2Fkafka-cluster.png?alt=media" alt=""><figcaption></figcaption></figure>

https://developers.redhat.com/learning/learn:apache-kafka:kafka-101/resource/resources:what-are-partitions

### **Kafka features**

<figure><img src="https://3925856803-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2FfJN93IwN0ptPSTmZ53eY%2Fuploads%2Fgit-blob-c609ec53b0e3eac005f2bfd099d912d3bcb4d551%2Fkafka-features.png?alt=media" alt=""><figcaption></figcaption></figure>

**Source Application (Kakfa Producer)**

* 데이터를 보내는 역할
  * ex. 클릭 로그, 결제 로그
* 여러 데이터 포맷을 지원(json, tsv, avro ..)

**Kafka**

* 각종 데이터를 담는 `토픽`을 가지고 있음
* 유연한 `Queue 역할`을 담당
* 데이터 흐름에 있어서 `fault tolerant`(고가용성)으로
  * 서버 이슈가 생기거나 갑작스럽게 전원이 내려갈 경우에서도 데이터를 손실 없이 복구 가능
* 낮은 지연(`latency`)과 높은 처리량(`Throughput`)을 통해 효과적으로 `대량의 데이터`를 처리

**Target Application (Kakfa Consumer)**

* 데이터를 가져가는 역할
* ex. 로그 적재, 로그 처리

***

## **카프카 클러스터 설치**

[AWS에 카프카 클러스터를 설치, 실행해보기!](https://www.youtube.com/watch?v=Qr0HVvtMFhg)

[AWS에 카프카 클러스터 설치하기(ec2, 3 brokers)](https://blog.voidmainvoid.net/325)

> 아파치 카프카를 설치하기 위해 2가지 애플리케이션이 필요

1️⃣ 주키퍼

* 카프카 관련 정보를 저장하는 역할

### 2️⃣ 카프카

## Producer

> 데이터를 카프카에 보내는 역할
>
> ex) 엄청난 양의 클릭 로그

**역할**

* Topic에 해당하는 메시지 생성
* 특정 Topic으로 데이터를 publish(전송)
* 카프카 브로커로 데이터 전송 시, 전송 성공 여부 확인 가능
  * 실패 시 재시도 가능

카프카 클라이언트인 컨슈머와 프로듀서 사용을 위해 아파치 카프카 라이브러리 추가

* ⚠️ 카프카는 브로커 버전과 클라이언트 버전의 하위호환성이 완벽하게 모든 버전에 대해서 지원하지 않으므로 주의 필요
  * 브로커 버전과 클라이언트 버전의 호환성 확인 필수

```groovy
// https://mvnrepository.com/artifact/org.apache.kafka/kafka-clients
implementation 'org.apache.kafka:kafka-clients:3.8.0'
```

카프카 프로듀서 작성 코드

* `bootstrap.servers`: 카프카 브로커의 주소 목록은 2개 이상의 ip, port 설정 권장
  * 하나의 브로커가 비정상일 경우 다른 브로커에 연결되어 사용 가능
* `key/value.serializer`: key/value 에 대한 StringSerializer 직렬화 설정
  * ByteArray, String, Integer Serializer 사용 가능
* topic: `click_log`
* value: `login`
* `Key`: 메시지를 보내면, 토픽의 파티션이 지정될 때 사용

<figure><img src="https://3925856803-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2FfJN93IwN0ptPSTmZ53eY%2Fuploads%2Fgit-blob-cfa5781515d81840ca24cdea49d8e3a6bc43dc43%2Fproducer.png?alt=media" alt=""><figcaption></figcaption></figure>

프로듀서가 토픽의 파티션에 들어가는 과정

*   key = null

    <figure><img src="https://3925856803-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2FfJN93IwN0ptPSTmZ53eY%2Fuploads%2Fgit-blob-8a12e24bd18ad25265aaffa941dc76b9502f7b6c%2Fproducer-key-null.png?alt=media" alt=""><figcaption></figcaption></figure>
*   key ≠ null

    * 키를 사용할 경우 파티션 개수를 고려
    * 추후에 생성하지 않는 것을 권장

    <figure><img src="https://3925856803-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2FfJN93IwN0ptPSTmZ53eY%2Fuploads%2Fgit-blob-bf6fd18d99eac96f8a67c95e76492025cb2e948a%2Fproducer-key-not-null.png?alt=media" alt=""><figcaption></figcaption></figure>

**데이터 유실, 브로커의 이슈에 대처하기 위한 추가 옵션과 코드가 필요**

참고.

* [아파치 Kafka 개요 및 설명](https://blog.voidmainvoid.net/179)
* [Kafka broker와 java client의 버젼 하위호환성 정리](https://blog.voidmainvoid.net/193)

***

## Consumer

> 다른 메시징 시스템에서는 컨슈머가 _**데이터를 가져가면 큐 내부 데이터가 사라지지만**_,
>
> 카프카에서는 _**컨슈머가 데이터를 가져가더라도 데이터 유지**_

이러한 특징은 카프카, 카프카 컨슈머를 데이터 파이프라인트로 운영하는데 핵심적인 역할

<figure><img src="https://3925856803-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2FfJN93IwN0ptPSTmZ53eY%2Fuploads%2Fgit-blob-52fb4d7fc6c24dad59f2ed1920c7a7de8019545b%2Fconsumer.png?alt=media" alt=""><figcaption></figcaption></figure>

* 카프카 컨슈머는 기본적으로 토픽 내부의 파티션에 저장된 데이터를 가져온다.
  * 이렇게 데이터를 가져오는 것을 `polling`

**역할**

* 토픽의 파티션으로부터 데이터를 폴링(polling)
  * 특정 DB에 저장하거나, 또 다른 파이프라인에 전달 가능
* 파티션 오프셋 위치 기록(commit)
* 컨슈머 그룹을 통해 병렬처리
  * 파티션 개수에 따라 컨슈머를 여러개 만들게 되면 병렬처리

**카프카 컨슈머 작성 코드**

* `bootstrap.servers`관련 설정은 프로듀서와 동일
* `group.id` (aka. 컨슈머 그룹) : 컨슈머들의 묶음
* `key/value.deserializer`: key/value 에 대한 StringDeserializer 역직렬화 설정
* consumer group: `click_log_greoup`
*   topic: `click_log` → consumer.subscribe()

    * 특정 토픽의 일부 파티션의 데이터만 가져올 경우 consumer.assign()
    * key가 존재하는 데이터라면 이 방식으로 데이터의 순서를 보장하는 데이터처리 가능

    ![](https://3925856803-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2FfJN93IwN0ptPSTmZ53eY%2Fuploads%2Fgit-blob-5fa831dc67e1ff76554a3bd876bfa3953dc1c74c%2Fconsumer-key.png?alt=media)
* 컨슈머는 poll()을 통해 데이터를 가져오는데, 설정 시간 동안 데이터를 대기
  * 0.5초 동안 데이터 도착을 대기하고 이후 코드를 실행
  * 0.5초 동안 데이터가 들어오지 않을 경우 빈값의 records 변수 반환
  * records 변수는 데이터 배치로서 레코드의 묶음 list
  * 실제 카프카에서 데이터를 처리할 때는 가장 작은 단위인 record로 나누어 처리

<figure><img src="https://3925856803-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2FfJN93IwN0ptPSTmZ53eY%2Fuploads%2Fgit-blob-58831813decc4285071b93912ab91ab807054ef3%2Fconsumer-code.png?alt=media" alt=""><figcaption></figcaption></figure>

**데이터가 컨슈머로 전달되는 과정**

* `offset`은 토픽, 파티션별 별개로 지정
* `offset`의 역할은 컨슈머가 _**데이터를 어느 지점까지 읽었는지 확인 용도**_
* 컨슈머가 데이터를 읽기 시작하면 `offset` `commit`
  * `offset` 정보는 `__consumer_offsets` 토픽에 저장
  * 컨슈머가 사고로 실행이 중지되어 재실행하면 _**시작 위치부터 다시 복구하여 데이터 처리**_ 가능

<figure><img src="https://3925856803-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2FfJN93IwN0ptPSTmZ53eY%2Fuploads%2Fgit-blob-0bfd782c462e70adad7d4f009f73bb46a754ad41%2Fconsumer-data.png?alt=media" alt=""><figcaption></figcaption></figure>

**컨슈머는 몇개까지 생성 가능❓**

* 컨슈머 1, 파티션 2 → 2개의 파티션에서 데이터 폴링
* 컨슈머 2, 파티션 2 → 각 컨슈머가 각각의 파티션을 할당

> 컨슈머 개수는 파티션 개수보다 작거나 같아야 한다.
>
> Consumer cnt ≤ partition cnt

<figure><img src="https://3925856803-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2FfJN93IwN0ptPSTmZ53eY%2Fuploads%2Fgit-blob-46162841da9713dab5f56063344e04bf41aaaf80%2Fconsumer-count.png?alt=media" alt=""><figcaption></figcaption></figure>

여러 파티션을 가진 토픽에 대해서 컨슈머를 병렬처리하고 싶을 경우,

* 반드시 컨슈머를 파티션 개수보다 적은 개수로 실행시켜야 한다.

**컨슈머 그룹이 다른 컨슈머들의 동작**

* 각기 다른 컨슈머 그룹에 속한 컨슈머들은 다른 컨슈머 그룹에 영향을 미치지 않음
* `__consumer_offsets` 토픽에는 컨슈머 그룹, 토픽별로 `offset`을 나누어 저장

> 이러한 카프카의 특징으로 하나의 토픽으로 들어온 데이터는 다양한 역할을 하는 컨슈머들이 각자 원하는 데이터로 처리가 가능

***

## Topic

> 카프카에는 다양한 데이터가 들어갈 수 있다.
>
> 여기서 데이터가 들어갈 수 있는 공간을 `Topic` 이라고 부른다.

<figure><img src="https://3925856803-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2FfJN93IwN0ptPSTmZ53eY%2Fuploads%2Fgit-blob-60e43e1fdbe9c595ed24f8fbe22bca4e360e6f18%2Fkafka-topic-1.png?alt=media" alt=""><figcaption></figcaption></figure>

일반적인 AMQP와는 다소 다르게 동작

* 카프카에서는 여러개의 토픽을 생성 가능
* 토픽은 데이터베이스 테이블이나 파일시스템의 폴더와 유사한 성질을 가짐
* 토픽은 이름을 가질 수 있는데 목적에 따라 설정
  * 어떤 데이터를 담는지 명확하게 명시

### 카프카 토픽의 내부

<figure><img src="https://3925856803-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2FfJN93IwN0ptPSTmZ53eY%2Fuploads%2Fgit-blob-9a532a8694fc04d5878f30d136111a20a456b12a%2Fkafka-topic-2.png?alt=media" alt=""><figcaption></figcaption></figure>

> 하나의 토픽은 여러개의 파티션으로 구성 가능

* 첫 번째 파티션은 0번 부터 시작
* 하나의 파티션은 큐같이 내부 데이터가 파티션 끝부터 쌓이게 된다.
* 컨슈머는 가장 오래된 순서로 데이터를 가져간다.
  * 데이터가 더이상 들어오지 않으면 컨슈머는 또 다른 데이터가 들어올 때까지 대기
* 컨슈머가 토픽 내부에서 데이터를 가져가더라도 데이터는 삭제되지 않는다.
  * 파티션에 남은 데이터는 새로운 컨슈머가 붙었을 때 다시 0번부터 사용 가능
  * 단, 컨슈머 그룹이 달라야 하고, `auto.offset.reset = earliest` 일 경우
  * **동일 데이터를 두 번 처리**할 수도 있는데 이는 카프카를 사용하는 아주 중요한 이유
    * 엘라스틱서치, 하둡 등 다른 저장소에 데이터를 저장해야 하는 등의 경우

**파티션이 2개 이상인 경우**

<figure><img src="https://3925856803-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2FfJN93IwN0ptPSTmZ53eY%2Fuploads%2Fgit-blob-736826d564e97be779d157c424d3d240bbfd8580%2Fkafka-topic-3.png?alt=media" alt=""><figcaption></figcaption></figure>

> 프로듀서는 데이터를 보낼 때 키 지정이 가능

* 키를 지정하지 않고(null), 기본 파티셔너설정을 사용할 경우 → 라운드 로빈(`Round robin`)으로 할당
* 키가 있고, 기본 파티셔너를 사용할 경우 → `키의 해시 값`을 구하고, 특정 파티션에 할당
* 파티션을 늘리는 것은 가능하지만 줄일 수는 없으므로 주의가 필요
  * 파티션을 늘리면 컨슈머의 개수를 늘려서 데이터 처리 분산이 가능
* 파티션 데이터의 삭제 타이밍은 옵션에 따라 다르다.
  * `log.retention.ms`: 데이터 최대 보존 시간
  * `log.retention.byte`: 데이터 최대 보존 크기

***

## **Broker, Replication, ISR**

<figure><img src="https://3925856803-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2FfJN93IwN0ptPSTmZ53eY%2Fuploads%2Fgit-blob-f26e5c0e8c7e7c7b699faa80d1823e7c80f187c2%2Fkafka-server.png?alt=media" alt=""><figcaption></figcaption></figure>

### Kafka **Broker**

> 카프카가 설치되어 있는 서버 단위

* 보통 `3개 이상의 브로커`로 구성하여 사용하는 것을 권장
* 만일, 파티션 1개, Replication 이 1 인 토픽이 존재, 브로커가 3대일 경우
  * 브로커 3대 중 1대에 해당 토픽의 정보가 저장

### Kafka **Replication**

> 파티션의 복제를 뜻함
>
> 파티션의 고가용성을 위해 사용

Replication

* `Replication = 1` 이라면, 파티션은 1개만 존재한다는 의미
* `Replication = 2` 이라면, 파티션은 원본 1개와 복제본 1개로 존재
* `Replication = 3` 이라면, 파티션은 원본 1개와 복제본 2개로 존재

<figure><img src="https://3925856803-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2FfJN93IwN0ptPSTmZ53eY%2Fuploads%2Fgit-blob-5031838cadc9db3a46c9acd0a8fca71fd4bdb5aa%2Fkafka-replication.png?alt=media" alt=""><figcaption></figcaption></figure>

**ISR(In Sync Replica)**

* 여기서 1개의 원본 파티션은 `Leader partition` 이라고 부르고,
* 나머지 복제본 파티션은 `Follower partition`
  * Leader partition 이 죽게 되면 복제본이 Leader partition 역할을 승계
* Leader + Follower partition 을 합쳐서 `ISR`(In Sync Replica) 이라고 볼 수 있다.

브로커 개수에 따라서 Replication 개수가 제한

* 브로커 개수가 3이면 Replication은 4가 될 수 없다.

### **Replication & Ack**

<figure><img src="https://3925856803-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2FfJN93IwN0ptPSTmZ53eY%2Fuploads%2Fgit-blob-ecd613301cf26298ead3960873b1c3df48be349d%2Freplication-ack.png?alt=media" alt=""><figcaption></figcaption></figure>

프로듀서가 토픽의 파티션에 데이터를 전달할 때 전달받는 주체가 `Leader partition`

* 프로듀서에는 ack 라는 상세 옵션을 통해 고가용성 유지 가능
* ack 옵션 중 하나를 선택
  * `0`
    * 프로듀서는 Leader partition에 데이터를 전송하고 응답값은 받지 않음
    * 데이터가 정상적으로 전송되었는지, 나머지 파티션에 정상 복제되었는지 보장할 수 없음
    * 속도는 빠르지만 데이터 유실 가능성 존재
  * `1`
    * 프로듀서는 Leader partition에 데이터를 전송하고, 데이터를 정상적으로 받았는지 응답값 수신
    * 단, 나머지 파티션에 복제되었는지는 알 수 없음
    * Leader partition이 데이터를 받은 즉시 브로커 장애가 발생한다면 데이터 유실 가능성 존재
  * `all`
    * 1 옵션에 추가로 follower partition에 복제가 잘 이루어졌는지 응답을 수신
    * 데이터가 유실될 일은 없지만, 속도가 0, 1 옵션에 비해 현저히 느리다는 단점

### **Replication count**

> replication 개수가 많아지면 그만큼 브로커의 리소스 사용량도 늘어나게 된다.

<figure><img src="https://3925856803-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2FfJN93IwN0ptPSTmZ53eY%2Fuploads%2Fgit-blob-2c7e4ba20928e86ef02f279001205ac74c27afbc%2Freplication-count.png?alt=media" alt=""><figcaption></figcaption></figure>

* 카프카에 들어오는 데이터량과 저장시간을 잘 고려하여 replication 개수 산정이 필요
  * 3개 이상의 브로커 사용 시 replication 은 3으로 설정하는 것을 추천

***

## Partitioner

> 프로듀서가 데이터를 보내면 무조건 파티셔너를 통해서 브로커로 데이터가 전송
>
> 파티셔너는 데이터를 토픽의 어떤 파티션에 넣을지 결정하는 역할

![](https://3925856803-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2FfJN93IwN0ptPSTmZ53eY%2Fuploads%2Fgit-blob-ad2ac9db8db5a341430ae6c3a796ca31a784cf7e%2Fpartitioner-1.png?alt=media)

**레코드에 포함된 메시지 키/값에 따라서 파티션의 위치가 결정**

* 프로듀서 사용 시 파티셔너를 따로 설정하지 않는다면 `UniformStickyPartitioner` 로 설정
  * 프로듀서에서 배치로 모을 수 있는 최대한의 레코드를 모아서 파티션으로 데이터를 전송
* Partitioner 인터페이스를 통해 커스텀 파티셔너를 생성할 수도 있음
  * 메시지 키/값, 토픽 이름에 따라 어느 파티션에 데이터를 보낼 것인지 설정 가능
  * ex. 특정 고객의 데이터를 더 빠르게 처리해주고 싶을 경우
    * 8개의 파티션에는 VIP 고객 데이터, 2개의 파티션에는 일반 고객 데이터

**메시지 키가 있을 때와 없을 때 다르게 동작**

*   메시지 키가 있는 경우

    * 파티셔너의 의해 특정한 해쉬값이 생성
    * 이 해쉬값을 기준으로 어느 파티션으로 들어갈지 선정
    * 동일한 메시지 키를 가진 레코드들은 동일한 파티션에 들어가므로 순서를 지켜서 데이터 처리할 수 있는 장점

    ![](https://3925856803-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2FfJN93IwN0ptPSTmZ53eY%2Fuploads%2Fgit-blob-28f1bd52e026b73324bd826aa75754dc73ec9b56%2Fpartitioner-2.png?alt=media)
* 메시지 키가 없는 경우
  * 라운드 로빈으로 파티션에 할당 (전통적인 라운드 로빈 방식과는 조금 다르게 동작)
  * 배치단위로 데이터 전송 시 파티션에 적절히 분배

***

## Consumer Lag

> 프로듀서가 넣은 데이터 오프셋과 컨슈머가 가져간 데이터의 오프셋
>
> 이 두 개의 오프셋 간의 차이가 `Consumer Lag`

![](https://3925856803-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2FfJN93IwN0ptPSTmZ53eY%2Fuploads%2Fgit-blob-f3f50cfa732c5457454ad29c255a48a5fd62694f%2Fconsumer-lag-1.png?alt=media)

`lag`은 적을 수도, 많을 수도 있는데,

* `lag` 를 통해 해당 토픽에 대해 파이프라인으로 연계되어 있는 프로듀서, 컨슈머의 상태에 대해 유추 가능
* 주로 컨슈머의 상태를 볼 때 사용

⭐️ `lag`은 각 파티션의 오프셋 기준으로 _**프로듀서가 넣은 데이터의 오프셋**_&#xACFC; _**컨슈머가 가져가는 데이터의 오프셋**_&#xC758; 차이를 기반

* ⭐️ 토픽에 여러 파티션이 존재할 경우 여러개의 `lag` 존재
  * ex) 컨슈머 그룹 1개, 파티션이 2개인 토픽에서 데이터를 가져간다면, `lag`은 2개가 측정
* 한 개의 토픽과 컨슈머 그룹에 대한 `lag`이 여러개 존재할 수 있을 때,
  * 그 중 높은 숫자의 `lag`를 `records-lag-max` 라고 부름.

![](https://3925856803-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2FfJN93IwN0ptPSTmZ53eY%2Fuploads%2Fgit-blob-3dd3bffac041c6d5dd9729261bb78236d23a8916%2Fconsumer-lag-2.png?alt=media)

***

## **Burrow**

[https://github.com/linkedin/Burrow](https://github.com/linkedin/Burrow)

> 카프카 컨슈머 Lag 모니터링 필수 요소

**Before**

`Kafka-client` 라이브러리를 사용해서 Java, Scala와 같은 언어를 통해 카프카 컨슈머 구현

* 구현한 `KafkaConsumer` 객체를 통해 현재 `lag` 정보를 가져올 수 있음
* `lag`을 실시간 모니터링하고 싶을 경우, 데이터를 `Elasticsearch`, `InfluxDB` 와 같은 저장소에 넣은 뒤
  * Grafana 대시보드를 통해 확인 가능
* 단, Consumer 단위에서 lag 을 모니터링하는 것은 위험하고 운영요소가 많이 들어감
  * 컨슈머 로직단에서 lag 을 수집하는 것은 컨슈머 상태에 디펜던시가 걸림
  * 컨슈머가 비정상적으로 종료되면 컨슈머는 lag 정보를 보낼 수 없으므로, 더이상 lag 측정 불가
  * 추가적으로 컨슈머가 개발될 때마다 해당 컨슈머에 lag 정보를 특정 저장소에 저장할 수 있도록 로직 개발 필요
  * 컨슈머 lag 수집 불가 컨슈머라면 모니터링이 어려워 운영이 까다로워짐

**Using Burrow**

> `Burrow`는 오픈소스로 Golang으로 작성

* 컨슈머 lag 모니터링을 도와주는 독립적인 애플리케이션

### Burrow의 특징

1️⃣ 멀티 카프카 클러스터 지원

*   카프카 클러스터가 여러개더라도 Burrow application 1개만 실행해서 연동하면

    * 카프카 클러스터들에 붙은 컨슈머의 lag을 모두 모니터링 가능

    ![](https://3925856803-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2FfJN93IwN0ptPSTmZ53eY%2Fuploads%2Fgit-blob-e2334a062307f754ea44f8e9c2fdfeebee8fd369%2Fburrow.png?alt=media)

2️⃣ Sliding window를 통한 consumer status 확인

* ERROR, WARNING, OK 로 표현
* `WARNING`: 데이터양이 일시적으로 많아지면서 컨슈머 오프셋이 증가되고 있을 경우
* `ERROR`: 데이터양이 많이지고 있는데 컨슈머가 데이터를 가져가지 않을 경우

3️⃣ HTTP api 제공

* 정보들을 Burrow API 를 통해 조회 가능

참고.

* [Burrow - kafka consumer의 지연(lag)을 모니터링할 수 있는 효과적인 opensource tool](https://blog.voidmainvoid.net/243)
* [Kafka Burrow에서 consumer의 lag을 정의하는(평가하는) 방법 - Consumer Lag Evaluation Rules](https://blog.voidmainvoid.net/244)
* [Kafka burrow http endpoint 정리](https://blog.voidmainvoid.net/245)

***

## RabbutMQ, Redis Queue, Kafka

메시징 플랫폼의 두 가지 종류

* 메시지 브로커: 이벤트 브로커 역할 수행 ❌
* 이벤트 브로커: 메시지 브로커 역할 수행 ⭕️

### 메시지 브로커

> **대규모 메시지 기반 미들웨어 아키텍처에서 사용**

* 미들웨어? 서비스하는 애플리케이션들의 아키텍처를 효율적으로 연결하는 소프트웨어
  * like) 메시징 플랫폼, 인증 플랫폼, 데이터베이스
* 메시지 브로커에 있는 큐에 데이터를 보내고 받는 프로듀서와 컨슈머를 통해 메시지를 통신하고 네트워크를 맺는 용도로 사용
* ex) `RabbutMQ`, `Redis Queue`

**메시지 브로커의 특징**

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

***

### 이벤트 브로커

> 메시지 브로커의 특징과 조금 다른 구조

1️⃣ 이벤트 또는 메시지라고도 불리는 레코드를 딱 하나만 보관하고, 인덱스를 통해 개별 엑세스를 관리

2️⃣ 업무상 필요한 시간동안 이벤트 보존 가능

* ex) `Kafka`, `AWS kinesis`

**이벤트 브로커의 특징**

* 딱 한번 일어난 이벤트 데이터를 브로커에 저장하여 단일 진실 공급원으로 사용
* 장애 발생 시 장애가 일어난 지점부터 재처리 가능

***

## Kafka Streams

> 토픽에 있는 데이터를 낮은 지연과 함께 안전하고, 빠른 속도로 데이터 처리

[Kafka Streams](https://kafka.apache.org/documentation/streams/)

**👍🏼 장점**

**1️⃣ 카프카와 완벽 호환**

> 메시지 유실, 중복이 발생하지 않고, 딱 한 번만 처리될 수 있는 강력한 기능을 제공

* 대부분 카프카를 이벤트 저장소로 사용하고, 저장된 데이터를 **Spark**, **Logstash** 같은 툴로 연동
  * 외부 오픈소스 툴은 빠르게 발전하는 카프카 버전에 따라오지 못 하는 단점이 존재
  * 반면, 스트림즈는 카프카 릴리즈 때마다 카프카 클러스터와 완벽하게 호환
* 카프카 보안, ACL 같은 것들이 붙어 있더라도 _**완벽 호환, 성능 개선도 진행**_

**2️⃣ 스케줄링 도구 불필요**

* 카프카와 연동하는 스트림 프로세싱 툴로 **Spark Streaming**이 널리 사용
  * 하지만, 스파크 운영을 위해 **yarn**, **messos** 같은 클러스터 관리자 또는 리소스 매니저 같은 것이 필요
  * 클러스터 운영을 위해 대규모 장비 구축이 필요
* 카프카 스트림즈를 사용하면 스케줄링 도구는 불필요
* **스트림즈 애플리케이션**은 컨슈머, WAS 애플리케이션을 배포하는 것처럼 _**원하는 만큼 배포 가능**_
  * 적은 양의 데이터를 처리한다면 2개 정도, 데이터를 많이 처리해야 한다면 스케일 아웃(10\~20개)

3️⃣ [Streams DSL](https://kafka.apache.org/21/documentation/streams/developer-guide/dsl-api.html)과 [Processor API](https://kafka.apache.org/10/documentation/streams/developer-guide/processor-api.html) 제공

* 스트림즈 구현은 대부분 **DSL**로 해결
  * 이벤트 기반 데이터 처리 시 필요한 다양한 기능들 map, join, window와 같은 메서드를 제공
  * **Streams DSL**만이 제공하는 KStream, KTable, GlobalKTable은 독특한 스트림 처리 개념
    * 카프카를 스트림 데이터 처리뿐 아니라 대규모 key-value 저장소로도 사용 가능
* **Streams DSL**에 없는 기능은 Processor API를 사용해서 로직을 작성
* 카프카의 풍부한 기능을 사용하고 싶다면 `Kafka Streams`가 답!

4️⃣ 로컬 상태 저장소를 사용

* 실시간으로 들어오는 데이터를 처리하는 방식
  * 비상태(`stateless`) 기반 처리
    * _**필터링**_&#xC774;나 _**데이터**_&#xB97C; 변환하는 처리
    * 데이터가 들어오는 즉시 바로 처리하고, 프로듀스하여 유실이나 중복이 발생할 염려가 적음
    * 개발을 위한 낮은 허들
  * 상태(`stateful`) 기반 처리
    * (분산 프로세스를) 직접 개발하기 위한 높은 허들
    * window, join, aggregation 같은 처리는 이전에 받은 데이터를 프로세스가 메모리에 저장하고 있으면서 다음 데이터를 참조해서 처리해야 함
* 상태 기반 처리 개발의 어려움을 극복하도록 도와주는 것이 `Kafka Streams`
  * 상태를 로컬 `rocksdb`를 사용해서 저장
  * 이 상태에 대한 변환 정보는 카프카 변경로그(`changelog`) 토픽에 저장
  * 프로세스에 장애가 발생하더라도 상태가 모두 안전하게 저장되어 자연스럽게 장애 복구

**Sample Code**

* payment 토픽에 메시지 키가 unknows 인 데이터를 필터링해서
* unknow-payment 토픽으로 보내는 스트림즈 코드

<figure><img src="https://3925856803-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2FfJN93IwN0ptPSTmZ53eY%2Fuploads%2Fgit-blob-1edd93cf8e323482b8b7b986ef67269047827b0e%2Fkafka-streams.png?alt=media" alt=""><figcaption></figcaption></figure>

***

## Kafka Connect

[Kafka Connect](https://docs.confluent.io/platform/current/connect/index.html)

> 반복적인 데이터 파이프 라인을 효과적으로 배포하고, 관리하는 _**데이터 파이프 라인 플랫폼**_

`Kafka Connect` 는 `Connect`와 `connector`로 구성

### Kafka Connect

> Connector를 동작하도록 실행해 주는 프로세스

파이프 라인으로 동작하는 Connector를 동작하기 위해 반드시 실행

.

`Connect`**는 크게 두 가지로 구성**

**1️⃣ 단일 실행 모드 커넥트**

* 간단한 데이터 파이프라인을 구성하거나 개발용으로 주로 사용

2️⃣ **분산 모드 커넥트**

* 여러개의 커넥트를 한 개의 클러스터로 묶어서 운영
* 클러스터로 묶은 커넥트는 일부 커넥트에서 장애가 발생하더라도 파이프라인을 자연스럽게 failover(장애 복구 기능)
  * 나머지 실행중인 커넥트에서 데이터를 지속적으로 처리할 수 있도록 지원

### Kafka Connector

> 실질적으로 데이터를 처리하는 코드가 담긴 _**jar 패키지**_

일련의 템플릿과 같은 특정 동작을 하는 코드 뭉치

* 커넥터 안에는 파이프라인에 필요한 _**여러 동작들과 설정, 실행 메서드**_&#xB4E4;이 포함
* 만일, 토픽에서 Oracle DB에 데이터를 저장하고 싶다면
  * 커넥터에 insert 메서드를 구현하고, 커넥터를 실행하는 방식으로 운영
* 직접 구현할 수도 있지만 깃허브, 컨플루언트에 오픈소스로 제공
  * 사용중인 카프카 클러스터와 싱크/소스로 사용하는 DB가 있다면 오픈소스 커넥터를 검색해서 사용 가능

.

**커넥터는 크게 두 가지로 구성**

**1️⃣ Sink Connector**

* 어딘가로 데이터를 싱크한다는 의미
* 특정 토픽에 있는 데이터를 Oracle, mySQL, ES 같이 특정 저장소에 저장을 하는 역할 (like. Consumer)
  * Oracle Sink Connector, mySQL Sink Connector, ..

**2️⃣ Source Connector**

* 데이터 베이스로부터 데이터를 가져와서 토픽에 넣는 역할 (like. Producer)

### Connect & Connector 관계

커넥트를 실행할 때 커넥터가 어디에 위치하는지 config 파일에 그 위치를 지정

*   커넥터 jar 패키지가 있는 디렉토리를 config 파일에 지정

    ```bash
    plugin.path=/var/connectors
    ```
* 그 후 커넥트를 실행하게 되면 이 jar 파일의 커넥터들을 함께 모아서 커넥터를 실행할 수 있도록 준비 상태에돌입
* 실행중인 커넥트에서 커넥터를 실행하려면 REST API 활용
  * 커넥트를 활용하면 파이프라인 생성 시 추가로 개발, 배포하는 과정 없이 REST API 를 통해 커넥터를 통한 파이프 라인들이 분산해서 생성
  * 예로, 특정 토픽에 있는 데이터를 Oracle DB의 특정 테이블로 보낼 때 JSON 으로 설정을 생성
    * 그리고 이 body를 REST API 를 통해 커넥트에 명령

<figure><img src="https://3925856803-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2FfJN93IwN0ptPSTmZ53eY%2Fuploads%2Fgit-blob-d8f76a05b4fe9d010cd2f21eb68d352458b7755d%2Fkafka-connect.png?alt=media" alt=""><figcaption></figcaption></figure>

> 개발, 배포, 모니터링 구축 과정들이 커넥트에서는 템플릿 형태로 커넥터를 개발하고,
>
> REST API를 통해 반복적으로 생성해서 뛰어난 효율을 가짐

_**반복적으로 파이프라인 생성이 필요할 때**_&#xB294; 컨슈머로 여러번 만들기보다 `커넥트`를 구축하여 `커넥터`를 반복적으로 실행하는 방식으로 진행하는 것을 권장

***

## **Confluent Kafka cloud Service**

[Kora by Confluent Cloud: 10x better Kafka](https://www.confluent.io/confluent-cloud/)

> 카프카를 직접 서버로 구축하지 않아도 SaaS 형태로 사용 가능한 서비스

카프카를 Saas 형태로 사용할 수 있는 서비스

* 1️⃣ AWS `MSK`(Managed Streaming for Kafka)
* 2️⃣ Confluent `Cloud Kafka`

**Confluent Kafka cloud Service 는 클릭 몇번의 설정으로 즉시 구축된 kafka cluster 사용 가능**

* 기존 개발자들이 직접 서버를 발급받고, zookeeper 설치 및 각종 script 명령을 통한 kafka 실행하는 방식과는 차원이 다른 방식
* 대부분의 동작이 웹 인터페이스를 통해 쉽고, 간편하게 사용 가능
* 클러스터 운영에 들어가는 개발자의 리소스를 줄일 수 있는 장점
  * 당장 개발자 리소스가 부족하고, 스트리밍 서비스를 빨리 만들어야 하는 경우 유용

***

## **빅데이터 플랫폼 아키텍처와 카프카**

초기에는 엔드 투 엔드로 각 서비스 애플리케이션으로부터 데이터를 배치로 모았음

* 데이터를 배치로 모으는 구조는 유연하지 못 할 뿐만 아니라
* 실시간 생성 데이터에 대한 인사이트를 서비스에 빠르게 전달하지 못하는 단점
* 원천 데이터로부터 파생된 데이터의 히스토리 파악이 어려움
* 계속되는 데이터의 가공으로 인해 데이터가 파편화 → 데이터 거버넌스를 지키기 어려움
  * 거버넌스: 빅데이터에 대한 체계적인 관리와 통제(프라이버시, 품질, 데이터 생명주기 ..)

<figure><img src="https://3925856803-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2FfJN93IwN0ptPSTmZ53eY%2Fuploads%2Fgit-blob-584fafc5675c1636ca52c748ae77f1b4dd2128f8%2Fend-to-end-architecture.png?alt=media" alt=""><figcaption></figcaption></figure>

### **Lambda Architecture**

> 위 단점들을 해소하고자 나온 아키텍처

람다 아키텍처는 세 가지 레이어로 나뉜다.

* 배치 레이어
  * 배치 데이터를 모아서 특정 시간, 타이밍마다 일괄 처리
  * 스파크 잡과 같은 대규모 배치 잡이 돌아가는 레이어
* 서빙 레이어
  * 가공된 데이터를 데이터 사용자, 서비스 애플리케이션이 사용할 수 있도록 데이터가 저장된 공간
  * 하둡이 대표적인 서빙 레이어 시스템
  * 대량의 데이터를 안정적으로 저장
* 스피드 레이어
  * 서비스에서 생성되는 원천 데이터를 실시간으로 분석하는 용도
  * 배치 데이터에 비해 낮은 지연
  * 카프카 같은 이벤트 스트리밍 플랫폼 위치한 레이어

<figure><img src="https://3925856803-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2FfJN93IwN0ptPSTmZ53eY%2Fuploads%2Fgit-blob-7661be148004d391dd87c4f41665f3ea7ee8e638%2Flambda-architecture.png?alt=media" alt=""><figcaption></figcaption></figure>

**⚠️ 한계점**

* 데이터를 배치 처리하는 레이어와 실시간 처리하는 레이어를 분리한 방식은 데이터 처리 방식을 명확히 나눌 수 있었지만 _**레이어가 결국 두 개로 나뉘기 때문에**_ 생기는 단점이 존재
  * 데이터 분석, 처리에 필요한 로직이 두 벌로 _**각각의 레이어에 따로 존재**_&#xD574;야 함
  * 배치 데이터와 실시간 데이터를 융합하여 처리할 때는 _**다소 유연하지 못한 파이프라인을 생성**_&#xD574;야 함
  * 두 번의 배포와 디버깅과 로깅, 모니터링도 각각 수행이 필요

### **Kappa Architecture**

> 람다 아키텍처의 단점을 해소하기 위해 제이 크렙스가 제안

람다 아키텍처의 단점으로 부각되었던 이슈(로직의 파편화, 디버깅/배포/운영 분리)를 제거하기 위해 배치 레이어를 제거하는 방안을 고려

* 스피드 레이어에서 모든 데이터를 처리할 수 있으므로 더욱 효율적으로 개발/운영을 진행

<figure><img src="https://3925856803-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2FfJN93IwN0ptPSTmZ53eY%2Fuploads%2Fgit-blob-39d6ec85732012e5f4b18cc0c9d706a99c8879d9%2Fkappa-architecture.png?alt=media" alt=""><figcaption></figcaption></figure>

### Streaming Data Lake

> 카파 아키텍처에서 서빙 레이어를 제거한 아키텍처

스피드 레이어로 사용되는 카프카에 분석과 프로세싱을 완료한 대용량의 데이터를 오랜 기간 저장하고 사용할 수 있다면 서빙 레이어는 제거되어도 된다는 가정으로 접근

* 기존에 사용하던 하둡을 카프카로 대체

<figure><img src="https://3925856803-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2FfJN93IwN0ptPSTmZ53eY%2Fuploads%2Fgit-blob-7e68e36661676d2a1301fb1e08f704fe7742a4ec%2Fstreaming-data-lake.png?alt=media" alt=""><figcaption></figcaption></figure>

카프카를 스트리밍 데이터 레이크로 사용하기 위해 개선해야 하는 부분

1️⃣ 스트리밍 데이터를 배치 데이터로 자유롭게 접근이 가능해야 한다.

* 스트리밍 데이터는 타임스탬프 기준으로 쿼리를 돌려 배치 형태로 데이터를 일부 뽑아낼 수 있지만 카프카로는 한계가 존재
* 카프카 스트림즈의 KTable, GlobalKTable로 구체화된 뷰를 통해 조회는 할 수 있지만, 스프림즈를 사용할 때만 가능한 부분
  * 확장성을 고려하면 아직 많은 개선이 필요

2️⃣ 자주 접근하지 않는 데이터를 비싼 자원에 유지할 필요가 없다는 점.

* 카프카 클러스터에서 자주 접근하지 않는 데이터는 오브젝트 스토리지와 같이 저렴하면서도 안전한 저장소에 옮겨 저장하고, 자주 사용하는 데이터만 브로커에서 사용하는 구분 작업이 필요

***

## Why Kafka?

데이터파이프라인 처리를 위해 redis, rabbitMQ 등 여러 애플리케이션들을 조합해서 사용해야 했다.

* 이런 애플리케이션 각각의 특징이 뚜렷하다보니 데이터 처리에 있어 _**파편화도 심각**_
* 파편화는 결국 유지보수의 어려움으로 직결되었고, 유지보수의 어려움은 비즈니스 성장에 영향

> 이러한 걸림돌을 타파하기 위해 만들어진 것이 `카프카`

* 카프카는 데이터 처리를 각각 여러 애플리케이션에서 처리하는 것이 아니라 중앙의 한 곳에서 처리할 수 있도록 `중앙집중화`한 것
* 카프카를 통해 웹사이트, 애플리케이션, 센서 등에서 취합한 데이터 스트림을 `한 곳에서 실시간으로 관리`
* 카프카는 대용량 데이터를 수집하고, 사용자들이 실시간 스트림으로 소비할 수 있게 만들어주는 일종의 `중추 신경`으로 작동

카프카가 많이 사용되고 있는 특징

**1️⃣ High throughput message capacity**

* 짧은 시간 내에 엄청난 양의 데이터를 컨슈머까지 전달
* 파티션을 통한 분산처리가 가능하므로 데이터 양이 많아질수록 컨슈머 개수를 늘려 병렬 처리가 가능
* 병렬 처리를 통해 데이터를 더욱 빠르게 처리

**2️⃣ Scalability and Fault tolerant**

* 카프카는 뛰어난 확장성
* 이미 사용되고 있는 카프카 브로커가 있더라도 신규 브로커 서버를 추가해서 수평 확장이 가능
* 여러 브로커 중 몇 대가 죽더라도 이미 replica로 복제된 데이터는 안전하게 보관되어 복구하여 처리 가능

**3️⃣ Undeleted log**

* 카프카 토픽에 들어간 데이터는 컨슈머가 데이터를 가지고 가더라도 데이터가 사라지지 않음
  * 만일 다른 플랫폼에서 이러한 처리를 하려고 했다면 복잡한 방식이 필요
* 카프카에서는 컨슈머의 그룹 아이디만 다르다면 동일한 데이터도 각각 다른 형태로 처리

아파치 카프카의 생태계는 꾸준히 발전

* 기본 개념인 컨슈머, 프로듀서 뿐만 아니라
* 카프카 스트림, 커넥터, ksqldb, avro, rest proxy 등
* 카프카에 적합한 여러 요소가 지속적으로 개발되고 발전
* 또한, fluentd, telegraf 등 여러 서드파티들에서도 카프카와의 연동을 정식으로 지원
