Select a result to preview

ksqlDB 101, part 1. 배경지식

#devlog #data_processing

이 글은 ksqlDB(당시에는 KSQL이라는 명칭이었습니다)를 학습하기 위해 정리한 연재글입니다.

이 글의 순서는 아래와 같습니다.

목차

  1. part 1. 배경지식
  2. part 2. Kafka Streams에 대해
  3. part 3. 실전 예시: 오픈소스를 통해 살펴보는 실시간 보안 이벤트 탐지 룰

1. ksqlDB를 알기 위해 필요한 것들

ksqlDB를 이해하고 효율적으로 사용하기 위해 아래 기능들에 대해 이해하고 넘어갈 필요가 있습니다.

사전 학습: 아파치 카프카의 API들에 대하여

아파치 카프카의 몇몇 API에 대한 설명을 살펴봅시다. 이 링크를 살펴본다면 ksqlDB를 살펴볼 때에도 도움이 될 것입니다.

너무 직역이라, 이를 풀어보자면 아래와 같습니다.

  1. 아파치 카프카의 프로듀서 API, 컨슈머 API는 말 그대로, 카프카 토픽을 프로듀스/컨슘 하는데 사용되는 네이티브 API 입니다.
  2. 스트림 API는, 한 토픽에서 다른 토픽으로 데이터의 흐름(streams of data)을 흘려보내는 기능을 의미합니다.
  3. 커넥트 API는 외부 시스템에서 카프카로 데이터를 뽑아오거나, 카프카에서 외부 시스템으로 데이터를 저장하는(sink data system이라고 일컫습니다) 기능을 의미합니다.

미리 말씀드리자면, ksqlDB는 이 스트림 API를 SQL문 작성 만으로 사용할 수 있도록 하는 애플리케이션이라 볼 수 있습니다.

Schema Registry

Avro란?

Avro는 스키마 정의를 아래의 방식으로 수행합니다.
• 파일타입
• 레코드의 길이
• 레코드의 이름
• 레코드 필드
• 필요한 데이터타입에 맞게 정의

아래와 같은 형식으로 담깁니다.

{"type": "typeName" ...attributes...}

e.g. document을 정의하고 namespace는 "Test"이며 이름은 "Employee"이고 2개의 필드를 가지고있다. 각각 string 타입의 "Name", int 타입의 "Age"이다.
라고 하는 데이터 타입에 대한 Avro 정의는 아래와 같습니다.

{
  "type": "record",
  "namespace": "Test",
  "name": "Employee",
  "fields": [
    { "name": "Name", "type": "string" },
    { "name": "Age", "type": "int" }
  ]
}

Avro의 primitive data type은 아래와 같습니다.

데이터타입 상세
null 아무값도 없는 값
int 32bit signed integer
long 64bit signed integer
float IEEE 754를 따르는 32-bit 부동소수점 값
double IEEE 754를 따르는 64-bit 부동소수점 값
bytes 8-bit unsigned bytes의 시퀀스
string 유니코드 캐릭터 시퀀스

Avro의 Complex data type은 다음과 같으며, 이는 "type": "<complex_data_type>" 와 같은 형태로 정의하며, 여기에서는 일부만 소개 하겠습니다.

데이터타입 상세
Record 다양한 애트리뷰트들의 컬렉션
Enum 아이템 리스트에 대한 enumuration
Arrays 배열 형식의 값을 담을 수 있는 데이터 타입
Maps key, value 쌍의 값을 담을 수 있는 데이터 타입

그 외의 내용은 아래 링크를 통해 확인해보시기 바랍니다.

Avro Serializer 를 어떻게 사용하나요?

자, 원하는 타입대로 직렬화/역직렬화를 수행할 수 있도록 포맷을 정했습니다. 그 다음은 어떤 절차를 거쳐야할까요? 아래 내용을 살펴봅시다. 이 내용은 파이썬의 confluent-kafka-python 라이브러리를 통해 사용하였음을 말씀드립니다.

  1. 프로듀서가 유효한 스키마를 가지고있는지 schema registry에 쿼리한다.
  2. schema registry는 스키마의 유효성에 따라 다르게 처리한다:
    1. 유효하지 않다면 KafkaAvroSerializer를 raise한다.
    2. 유효하면 스키마 ID를 메시지에 추가하고 카프카에 쏜다.
  3. 필요에 따라 produce/consume을 한다(이 내용은 상세히 봅시다!)

Produce 과정

이 예시코드를 살펴보시면 이해가 빠릅니다!

  1. AvroSerializer 객체를 통해 Schema Registry 정보를 받고 직렬화/역직렬화 처리준비를 완료함
  2. 카프카 정보를 받은 후 토픽에 produce 수행

Consume 과정

이 예시코드를 살펴보시면 이해가 빠릅니다!

  1. AvroDeserializer 객체를 통해 Schema Registry 정보를 받고 직렬화/역직렬화 처리준비를 완료함
  2. 카프카 정보를 받은 후 토픽으로부터 subscribe 한다.
  3. 값이 있으면 그때부터 계속 poll하면서 가져온다.

스키마 레지스트리의 작동 원리

스키마 레지스트리 호출을 언급했는데, 정확히 어떤식으로 작동하는 걸까요? 그 내용에 대해선 아래 도식을 통해 설명하겠습니다.

  1. Producer는 Avro포맷으로 REST API에 POST 를 수행합니다.
  2. REST interface는 Schama Registry에 스키마를 전송합니다.
    1. 그 후 카프카에 binary data를 전송합니다.
    2. 이때 binary data 에는 schema ID가 포함되어 있습니다.
  3. Consumer는 binary data를 Kafka에서 consume합니다.
    1. 그 후 해당 정보를 가지고 스키마 레지스트리(혹은 로컬캐시)에서 스키마 정보를 탐색후 가져오고, 이를 이용하여 역직렬화합니다.

스키마 레지스트리의 작동방식

스키마 레지스트리 서버세팅 (Docker Compose 사용 시)

추가로 확인해야할 설정값(SSL, SASL 등의 설정이 필요하다면 아래 링크를 참조해 주세요)

deployment시 점검해야할 스펙은 아래와 같습니다:

서버 세팅이 완료되면, 그 후 스키마 레지스트리 서버에 대해 API 요청을 테스트해볼 수 있습니다:

스키마 레지스트리의 장단점

schema를 사용할 때 생기는 장점

schema를 사용할 때 생기는 단점

스키마 레지스트리를 강요하지하지는 않지만, 많은 케이스 에서 스키마 레지스트리 형식으로 포맷을 보장하기를 바라는 것으로 보입니다.

마무리

이번 글을 통해, 아래 내용들을 살펴볼 수 있었습니다:

  1. 아파치 카프카의 Produce/Consume 뿐 아니라 Streams, Connect 기능을 이용하여 데이터의 흐름을 추가적으로 제어할 수 있음을 알 수 있었습니다.
  2. 데이터 흐름이 복잡해짐에 따라 데이터 직렬화/역직렬화에 대한 관리도구 및 직렬화/역직렬화 도구가 대두되었습니다. 이는 각각 스키마 레지스트리 및 Apache Avro를 말합니다.
  3. 스키마 레지스트리와 Apache Avro에 대해 간략하게나마 살펴보고, 장단점을 살펴보았습니다.

다음 파트에선 ksqlDB를 명확히 이해하기 위해 Kafka Streams에 대해 명확히 알아보도록 합시다.

읽어주셔서 대단히 감사합니다.