Kafka

Kafka는 요즘 Uber, 에어비엔비, 넷플릭스 등 많은 회사에서 사용하는 빅데이터 공간에서 가장 인기있는 이벤트 스트리밍 플랫폼이다.

Event Streaming Platform

기존의 Enterprise Architecture에서는 모든 앱이 사일로화(부서, 사업, 브랜치별로 데이터 단절)되어 실시간으로 이상을 통합하거나 감지하는 것은 물론 실시간 비즈니스 의사결정을 내리는 것이 매우 어려웠다. 또한 데이터베이스를 기준으로 수많은 임시 솔루션들이 나오고, 이러한 데이터베이스 상호 연결은 아래와 같이 얽히고 얽힌 지점 간 통합으로 이어졌다.

이를 극복하고자, 중앙 집중식 허브에 연결하여 실시간 정보를 분석하고 이슈가 발생하는 즉시 원하는 대로 변환할 수 있는 이벤트 스트리밍 플랫폼이 등장했다.

이벤트 스트리밍 플랫폼은 데이터 저장소, 로그, 타사 앱 등의 모든 엔터프라이즈 애플리케이션에 대한 단일 접점의 역할을 한다. 대표적인 대규모 스트리밍 이벤트 처리 엔진으로는 Kafka, Spark Streaming, Apache Flink 등이 있다.이러한 엔진들을 통하여 실시간 이벤트들을 모은다. 이러한 이벤트들은 카메라와 같은 스트림 처리 엔진을 통해 모니터링 및 기록되며, 실시간으로 이러한 이벤트를 재생하고 개입하고 반응할 수 있다.

장점

1. 단일 솔루션으로 애플리케이션과 데이터 인프라를 볼 수 있다.(높은 처리량, 지속성, 짧은 대기시간, 모든 이벤트 및 시스템이 연결되는 범용 이벤트 파이프라인)

   -> 이로 인해 서로 다른 시스템끼리 서로 커뮤니케이션이 가능해졌다.

2. 레거시 데이터 웨어하우스, RDBMS를 현대화하고, 탄력적인 상태 저장 데이터 저장소를 구축 가능

3. 하이브리드 클라우드 스트리밍 플랫폼 구축

Kafka의 구성요소

Kafka : Pub/Sub 모델

아파치 카프카는 중앙집중형 메시지 관리 방식으로 메시지의 생성과 소비와 관리를 완전히 독립시킨 pub-sub 메시징 시스템이다. 메시지를 관리하는 데이터 큐를 중앙에 두고, 독립적인 데이터의 제공, 소비가 이루어 진다. 이와 같은 패턴을 publish / subscribe 모델 패턴이라고 하며, 이런 느슨한 결합을 통하여 제공과 소비를 요청한 곳에서 장애가 발생할 때에도 의존성이 없으므로 안정적인 데이터 처리가 가능하다.

Kafka Cluster

카프카 클러스터는 임의 갯수의 노드로 구성되고, 여기서 노드를 broker라고 바꿔 말할 수 있다. 보통 클러스터에는 1개 이상의 브로커를 두어 특정 노드의 물리적 장애로 인한 데이터 유실을 대비한다.

Broker

단일 Kafka 서버를 Broker라고 한다. Broker는 Producer로부터 메시지를 수신하고 Offset을 할당하고 디스크의 저장소에 메시지를 커밋한다. Consumer에게 서비스를 제공하여 파티션에 대한 Fetch 요청에 응답하고 디스크에 커밋된 메시지에 응답한다. 하드웨어 성능에 따라 단일 Broker가 초당 수천 개의 파티션과 수백만 개의 메시지를 쉽게 처리할 수 있다.

Topic & Partition

Topic은 메시지를 받을 수 있는 논리적인 모델로 데이터를 구분하기 위한 단위이다. 파일 시스템의 데이터베이스 테이블 또는 폴더와 같다. 데이터의 처리량을 높이기 위해 1개의 Topic은 1개 이상의 파티션을 가지고 아래와 같이 데이터를 분산시켜 저장한다.

3개의 Broker로 구성된 Kafka Cluster에 Topic을 4개의 파티션, replica는 3개로 설정해서 생성하였다. 여기서 replica를 설정하는 이유는 브로커 노드의 장애나 Fail에 대비하기 위해서 이다. 그리하여 0~3까지의 파티션이 생성되고, 각 파티션이 3개로 복제가 되었다.

여기서 각 파티션마다 리더가 선출되고, 나머지는 팔로워가 된다. 리더는 Producer, Consumer를 통한 데이터 write/read를 수행하고, 팔로워는 데이터 복제만을 담당한다. 만일 리더노드가 있는 브로커에 장애가 발생한다면 zookeeper에서 나머지 팔로워 중에 새로운 리더를 선출하여 데이터의 유실을 방지하고 복구를 진행할 수 있도록 한다. 따라서 위 경우는 한 replica가 3이므로 2번의 장애까지 견딜 수 있는 것이다.

하지만 무조건 파티션의 수를 늘리는 것만이 좋은것은 아니다. 파티션이 많아지면 Kafka가 참조할 자원들이 많아져 리소스가 낭비될 수 있다. 또한 장애 복구 시간이 증가한다. 리더가 있는 브로커 장애시 파티션 수가 증가함에 따라 리더 선출에 대한 시간이 비례하여 상승한다.

특히 Kafka에서는 파티션을 증가하는 것은 아무때나 변경이 가능하지만, 반대로 파티션의 수를 줄이려면 토픽을 삭제하는 방법밖에 없다. 또한 메시지는 HDFS와 같은 append-only 방식으로 토픽에 기록된다.

위 그림은 Producer와 Broker, Consumer 의 구조를 그림으로 표현한 것이다. Producer는 1~9까지 순서대로 메시지를 publish하였다. Consumer는 각 파티션마다 가지고있는 Offset부터 차례대로 메시지를 읽어온다. 만약 하나의 Consumer가 3개의 파티션을 모두 subscribe한 경우엔, Producer가 보냈을때의 순서대로 오지 않을 수 있다. 다른 파티션끼리는 순서를 보장하지 않기 때문이다. 하지만 하나의 파티션 내에서는 메시지의 순서가 보장된다. 그래서 만약 메시지의 순서가 중요한 경우, 파티션의 수를 1로 두거나, 순서가 보장되어야 하는 메세지는 같은 파티션에 들어가도록 파티션키(해시키)를 설정하여야 한다. 하지만 싱글 파티션으로 했을때 문제가 발생할 수 있다는 이론들이 있다. 해당 브로커가 죽었다가 다시 살아나는 그 찰나가 문제이다. 물론 옵션을 통해서 죽었던 곳부터 다시 소비를 하는 offset.reset 전략을 택했지만, 네트워크 이슈로 인해 중복소비되는 상황이 발생할 수도 있다는 것이다. confluent 팀도 이러한 문제점을 고민하고 나온것이 Exactly-Once-Delivery라는 컨셉이다.

Producer

topic에 데이터를 쓰는 역할, 새로운 메시지를 만든다. 일반적으로는 메시지는 특정 토픽에 대해 생성되며, 모든 파티션에 균등하게 메시지 균형을 유지한다. Producer가 해시키를 이용하여 특정 파티션으로 메시지를 보낼 수도 있다.

Consumer

topic에서 데이터를 가져가는 역할. 메시지를 읽는다. 특정 Topic을 구독함으로써, 자기가 스스로 조절해가면서 메시지를 받을 수 있는 것이다. 해당 topic 내의 각 파티션에 존재하는 Offset과 Consumer가 이전에 소비했던 Offset의 위치를 통해 마지막으로 읽었던 위치에서부터 다시 읽어들일 수 있다.

Consumer Group

말 그대로 Consumer들의 묶음이다. 기존의 Message Queue 솔루션에서는 Consumer가 메시지를 가져가면, 해당 메시지는 큐에서 삭제된다. 하지만 Kafka는, Consumer가 메시지를 가져가도 큐에서 즉시 삭제되지 않으며, 하나의 토픽에 여러 Consumer Group이 붙어 메시지를 가져갈 수 있다.(2.x 기준 default 7일) 아래 그림과 같이 구성되어 있다.

여기에 기본적인 룰 하나가 존재한다. 반드시 토픽의 파티션과 Consumer Group은 1:n이여야 한다. 위 그림을 보자. 그림에서 보다시피, 단 하나의 파티션에는 동일한 Consumer Group에서 반드시 하나의 Consumer만 할당이 된다는 것이다.

하지만 아래 그림과 같이 파티션과 Consumer의 비율이 2:3인 경우, 하나의 Consumer는 할당된 파티션 없이 대기만 할 것이다.

이와 같은 상황을 방지하기 위하여 파티션과 Consumer 수도 고려해야 한다. 보통은 개수를 같이 맞춰주는 것을 권장하지만, 실제 메세지가 쌓이는 속도보다 처리하는 속도가 훨씬 빠르다면 굳이 1:1 매핑보다는 파티션개수 => 컨슈머개수로 설정하는 것도 나쁘지 않다.

그렇다면 Consumer Group이 존재하는 이유는 무엇일까? 기본적인 룰이라고 말했던 파티션과 Consumer Group의 1:n 비율을 보면 알 수 있다. Consumer Group은 하나의 토픽의 파티션에 대한 별도의 Offset을 관리한다. Consumer가 다운된다면, 해당 Consumer가 속해 있던 그룹에서 rebalancing을 하여 그룹내의 다른 Consumer에 파티션을 할당하게 되어 그전에 Consumer의 소비를 이어서 하게 된다. 그룹 내에서 Offset의 정보를 서로 공유하고 있기 때문에, 다운 직전의 Offset 위치를 알고 그 다음부터 소비하면 문제가 없어지는 것이다.

Offset

지속적으로 증가하는 정수 값을 가진 메타데이터. 배열에서의 인덱스과 같다. 파티션의 각 메시지에는 고유한 오프셋이 있다. 이 오프셋으로 Consumer는 순서를 찾고 다음 가져올 데이터를 찾을 수 있다.

Commit Log

저장된 레코드나 임시 메시지로 처리되지 않고 지속적으로 업데이트되는 Event 스트림으로 처리되는 지속 파티션. 이러한 스트림은 쉽게 액세스할 수 있고, Event가 발생하는 즉시 처리할수 있을만큼 빠르며, 이전에 발생한 Event를 저장할 수 있다.

Zookeeper

카프카 아키텍처 내에서 카프카 브로커들을 하나의 클러스터로 코디네이팅하는 역할을 수행. 좀더 자세히 설명하자면, 카프카 클러스터의 설정 정보를 최신으로 유지하며, 클러스터 내에서 브로커 서버가 추가되거나 삭제될 때 그 정보를 클러스터에 속한 서버들끼리 공유할 수 있도록 한다. 또한 브로커 서버는 주키퍼에게 토픽 메타데이터를 공유한다.

이전 카프카 버전에서는 주키퍼가 직접적인 데이터 관리에도 참여했었다. Consumer측에서 읽은 파티션의 Offset 정보를 주키퍼에게 공유했었는데, 현재는 이 Offset 정보를 Consumer group에서 관리한다.

Kafka Reference Architecture

다음 그림은 Kafka의 레퍼런스들의 아키텍처이다.

REST Proxy

confluent에서는 REST Proxy라고 불리는 카프카 클러스터를 위한 RESTful interface application을 오픈소스로 제공하고 있다. Kafka 클라이언트를 사용하지 않고 범용적으로 사용되는 http를 사용하여 데이터를 넣고 뺄 수 있다는 점이 특징이다.

Schema Registry

RESTful 인터페이스를 사용하여 스키마를 관리하거나 조회하는 기능을 제공. 이때 Avro 포맷을 이용한다. Avro 포맷은 아래 그림과 같이 나눠져 스키마는 스키마 레지스트리로, 실데이터는 Kafka로 등록된다.

잘못된 스키마의 메시지를 전달한다면 스키마 레지스트리에 등록하는 과정에서 실패고 Kafka에 실데이터가 전달되지 않게 된다. 그리하여 Consumer에서는 잘못된 메시지를 받을 일이 없어 Consumer에서 잘못된 스키마로 인한 에러는 발생하지 않는다. 또한 Kafka에 전달되는 바이너리 데이터에 스키마가 포함되지 않기 때문에 상대적으로 적은 용량의 데이터가 전달된다.

하지만 Schema Registry의 역할이 굉장이 중요하여 Schema Registry의 장애 발생 시 정상적인 메시지 전달이 불가하다.

Kafka Connect

Kafka Connect는 Kafka 브로커와 다른 어플리케이션간에 데이터 핸들링을 위한 플러그형 프레임워크이다. Connector를 이용하여 데이터 소스와 Kafka를 연결해준다. 다시말해, Kafka Connector가 Producer와 Consumer 역할을 한다고 보면 된다. Producer 역할을 하는 Kafka Connector를 Source Connector라고 하고, Consumer 역할을 하는 Kafka Connector를 Sink Connector라고 한다.

Kafka Connect에서는 Kafka를 여러 Source 및 Sink와 통합하여 자체 커넥터를 개발할 수 있는 프레임워크를 제공한다.

KSQL

Kafka에 대한 스트림 처리 작업을 수행하기 위해 사용하기 쉽고 강력한 SQL 인터페이스를 제공한다.

Kafka Streams

실시간으로 이벤트 스트림을 처리하고 강화할 수 있는 stateful 및 stateless 애플리케이션을 구축하기 위한 프레임워크이다..

Confluent Control Center

Kafka 브로커의 상태를 확인할 수 있으며, Kafka Connect, KSQL 및 Schema Registry와 같은 Confluent Platform의 구성 요소에 유용한 통합 GUI를 제공한다. 보통 Control Center는 K8S에 배포되므로, provisioning 자동화, kafka 클러스터 재조정, self healing, auto scaling과 같은 K8S의 모든 이점이 Control Center의 핵심 구성 요소가 되었다.

Kafka Streams

Kafka의 정의는 분산 이벤트 스트리밍 플랫폼으로써 Producer와 Consumer를 통해 데이터를 생산하고 받아와서 처리하는 곳에 사용해 왔다. 이때 Consumer로 데이터를 받아와서 처리하는 것보다 더 빠르고 안전하게 실시간으로 처리할 수 있게 Kafka 에서 지원해준 것이 Kafka Streams 이다. 더욱 자세하게는, 어떤 Topic으로 들어온 데이터를 Consume하여 Kafka Streams에서 제공하는 처리 로직을 통해 처리 후 다른 Topic으로 전송하거나 끝내는 부분을 수행하는 Confluent 공식 지원 자바 라이브러리이다.

구조

Kafka Streams를 사용하기 위해서는 데이터의 흐름을 표현하기 위해 노드와 선으로 이루어진 Topology라는 개념을 알아야 한다.

Topology는 ‘노드’와 ‘노드를 이은 선’으로 이루어져 있다. Kafka Streams를 도식화해 보면 아래와 같다.

Source Processor

Topology의 시작 노드이며, 데이터를 처리하기 위해 최초로 선언해야 하는 노드이다. Kafka와 연결된 프로세서이며, 하나 이상의 토픽에서 데이터를 가져오는 역할을 한다.

Stream Processor

다른 프로세서가 반환한 데이터를 처리하는 역할

Sink Processor

Topology의 마지막 노드이며, 데이터 전달을 위해 마지막에 선언해야 하는 노드이다.

Kafka와 연결된 프로세서이며, 데이터를 Kafka의 특정 토픽으로 저장하는 역할을 한다.

특징

카프카와 완벽 호환

Confluent 공식 지원 라이브러리이기 때문에 스파크, 로그스태시와 같은 외부 오픈소스들과 다르게 카프카 버전에 맞춰서 라이브러리를 제공한다.

스케줄링 도구가 필요 없다.

카프카와 연동하는 스트림 프로세싱 툴로 가장 많이 사용하는 것이 Spark Stream이다. 이를 통해 마이크로 배치 처리를 하는 이벤트 데이터 애플리케이션을 만들 수 있다. 하지만 이를 운영하기 위해선 yarn과 같은 클러스터 관리자, 리소스 매니저와 같은 것과 대규모 장비 등이 필요하다. 반면 Spark Streams를 사용하면 스케줄링 도구가 필요 없다. 자신이 원하는 만큼 적은 양의 데이터면 2개, 대용량이면 10~20개의 애플리케이션을 WAS 애플리케이션 배포하듯이 배포하면 된다.

스트림즈DSL과 프로세서API를 제공

데이터 처리를 할때 스트림즈 DSL에서 map, join, window와 같은 메서드들을 제공한다. 또한 스트림즈DSL만이 제공하는 KStream, KTable, GlobalKTable은 카프카를 스트림 데이터 처리뿐만 아니라 대규모 key-value 저장소로도 사용할 수 있는 기능을 가지고 있다.

• KStream

  데이터를 조회하면 토픽에 존재하는 모든 데이터 출력하며 key의 중복 허용

• KTable

  데이터를 조회하면 토픽에 존재하는 모든 데이터를 가져오면서 동일한 key의 데이터가 있으면, 최신 데이터로 덮어씌운다.

• GlobalKTable

  KTable로 선언된 토픽은 1개 파티션이 1개 task에 할당되지만, GlobalKTable로 선언된 토픽은 모든 파티션 데이터가 각 task에 할당되어 사용되는 차이점 존재

  KStream과 KTable 조인 해결 가능

자체 로컬 상태 저장소를 사용한다.

실시간으로 들어오는 데이터를 처리하는 방식은 비상태기반(Stateless) 처리, 상태기반(Stateful) 처리 두가지가 있다. 비상태기반 처리는 데이터가 들어올 때마다 Produce하면 되기 때문에 데이터의 유실이나 중복 발생 염려가 적고 개발이 쉽다. 상태기반 처리는 직접 구현이 어렵다. window, join, aggregation과 같은 처리는 이전에 받았던 데이터를 프로세스가 메모리에 저장하고 있으면서 다음 데이터를 참조해서 처리해야 한다.

이때 Kafka Streams는 로컬에 rocksdb를 사용해서 상태를 저장하고, 이 상태에 대한 변환 정보는 Kafka의 changelog 토픽에 저장한다. 그리하여 프로세스에 장애가 발생하더라도 그 상태가 모두 안전하게 저장되기 때문에 자연스럽게 장애 복구가 될 수 있다.

KSQL / KSQLDB

지속적인 스트림 형태로 들어오는 데이터에서 SQL쿼리의 특정 키 값으로 조회하려는 것은 불가능하다. 이걸 가능하게 하는 것이 KSQL 이다. KSQL은 Kafka Streams를 보완하기 위해 Kafka에 대해 스트림 처리 작업을 수행하는 강력한 SQL 인터페이스를 제공한다. 구성은 아래와 같다.

장점

• KSQL은 내부적으로 Kafka Streams를 기반으로 컴파일되며, 실시간으로 들어오는 데이터에서 SQL 쿼리문을 통하여 Error 혹은 원하는 데이터를 뽑아낼 수 있어 빠르게 관찰하고 대응해야 하는 이벤트 기반 마이크로서비스에 이상적이다.
• 유사 SQL문을 이용하여 로직을 구성하기 때문에 프로그래밍 지식이 많이 요구되지 않는다.
• 구현된 KSQL문을 서버에 배포하면 엔진 내에서 맞춰 구동되기 때문에, 사용자가 직접 서버 배포까지 고려할 필요가 없다.

위와 같은 특징 때문에 카프카 토픽에 대한 데이터 분석에 자주 이용된다.

예전에는 아래 그림과 같이 가운데 KSQL 처리를 하는 Cluster가 따로 있다. 그리하여 JVM에서 running을 하여 Kafka Cluster에 쿼리를 수행한다.

위와 같이 하면 JVM 클러스터를 따로 관리를 해야한다. 이에 한단계 더 발전한 것이 KSQLDB이다.

KSQLDB의 특징

특징1 : Pull query

기존 RDB에서 데이터를 가져오는 것을 Pull query 라고 한다. 하지만 KSQL에서는 지속적으로 변화하는 스트림에서 데이터를 밀어낸다는 의미로 push quries라는 명칭이 붙게 되었다. 이러한 Push query는 영구적으로 실행되며 새로운 변경이 발생할 때마다 결과를 업데이트 해준다. 과거 버전의 KSQL에서는 RDB와 같이 현재 상태를 가져오는 Pull query에 대한 기능이 없었다. 하지만 KSQLDB에서 EMIT CHANGE라는 것이 추가되면서 RDB와 같이 사용할 수도 있게 되었다. 기존 Kafka가 가지고 있던 저장소(DB)로써의 부분을 더욱 효과적으로 보완하고자 나온 것이다.

아래는 운전자의 위치를 기록하는 앱의 쿼리문이다. 위치를 변경해 가면서 현재 시점의 특정 운전자의 위치를 얻으려고 할때, KSQLDB에서는 아래와 같이 Pull query를 날릴 수 있다.

SELECT ride_id, current_latitude, current_longitude
FROM ride_locations
WHERE ROWKEY = ‘6fd0fcdb’;

+-----------+-----------------------+-----------------------+
|RIDE_ID    |CURRENT_LATITUDE       |CURRENT_LONGITUDE      |
+-----------+-----------------------+-----------------------+
|45334      |37.7749                |122.4194               |
+-----------+-----------------------+-----------------------+

여기서 EMIT CHANGES를 추가하면 현재 운전자에 대한 위치 변경 기록을 지속적으로 받을 수 있다.

SELECT ride_id, current_latitude, current_longitude
FROM ride_locations
WHERE ROWKEY = ‘6fd0fcdb’
EMIT CHANGES;

+-----------+-----------------------+-----------------------+
|RIDE_ID    |CURRENT_LATITUDE       |CURRENT_LONGITUDE      |
+-----------+-----------------------+-----------------------+
|45334      |37.7749                |122.4194               |
|45334      |37.7749                |122.4192               |
|45334      |37.7747                |122.4190               |
|45334      |37.7748                |122.4188               |
+-----------+-----------------------+-----------------------+

여기서 Pull query의 성능은 대용량 데이터에 따라서는 RocksDB의 성능에 따라간다.

특징2 : Connector Management

KSQL은 Kafka Cluster 밖에 있었다. 그래서 어떤 데이터를 가져오지에 대해서 Kafka Cluster 내에 따로 설정을 해주었어야 했다. 하지만 KSQLDB는 설정을 따로 할 필요 없이 쿼리문 자체에서 만들 수 있게 되었다. 쿼리문은 아래와 같다.

CREATE SOURCE CONNECTOR rider_profiles WITH (
      'connector.class'          = 'io.confluent.connect.jdbc.JdbcSourceConnector',
      'connection.url'           = 'jdbc:postgresql://postgres:5432/postgres',
      'key'                      = 'profile_id',
      ...
);

이것이 KSQLDB를 개발한 근본적인 이유이다. Confluent에서는 스트림 데이터를 처리할 때 만들어야 하는 시스템의 복잡성 때문이라고 설명하고 있다.

아래는 기존 KSQL의 ETL 동작을 구조화한 그림이다.

왼쪽부터 순서대로 데이터 추출 - Kafka 저장 - 데이터 변형 - 데이터 로드 - APP에 데이터 저장 으로 동작하게 된다. 이 과정에서 여러 시스템들을 각각 관리하는데 부담이 된다는 것이 요점이다. KSQLDB에서는 이걸 RDB를 통한 CRUD 만으로 구조를 간단히 할 수 있다고 한다. 구조는 아래와 같다.

아예 KSQLDB 내에서 어떤 곳에서 어떤 데이터를 가져올 것이다 등의 모든 과정을 한방에 처리할 수 있게 되었다. KSQL의 ETL 동작 구조에 있었던 Stream Processing, Connectors 이런 것들을 KSQLDB에서 처리할 수 있도록 한 것이다.