카프카 스트림즈

[manjeong-dev]

토픽에 적재된 데이터를 실시간으로 변환하여 다른 토픽에 적재하는 라이브러리

카프카 공식 라이브러리

스트림즈 애플리케이션

• 카프카 브로커의 장애가 발생하더라도 exactly once

• 장애 허용 시스템(falut tolerant system) 가지고 있음

• 데이터 처리 안정성이 매우 뛰어남

• 스트림즈DSL 제공, 필요하다면 프로세서 API 사용하여 기능 확장

• 소스토픽(사용하는 토픽)과 싱크 토픽(저장하는 토픽)의 카프카 클러스터가 서로 다른 경우 > 스트림즈 지원 x > 컨슈머와 프로듀서 조합으로 직접 클러스터를 지정하는 방식으로 개발

• 스트림즈 애플리케이션 : 내부적으로 스레드를 1개 이상 생성할 수 있음

• 스레드는 1개 이상의 task를 가짐

• task

  • 스트림즈 애플리케이션을 실행하면 생기는 데이터 처리 최소 단위

• 컨슈머 스레드 늘려 > 병렬 처리 | 파티션&스트림즈 스레드(또는 프로세스) 개수를 늘림 > 병렬 처리

  

• 2개 이상의 서버로 구성 가능

• application.id 를 지정하여 동일한 처리로 묶을 수 있음

• 태스크0의 애플리케이션이 장애가 나도 다른 애플리케이션으로 파티션 연결을 옮겨 fail over 가능(application.id로 묶어서)

• 멀티 스레드, 멀티 프로세스로 처리 가능

topology

• 2개 이상의 노드들과 선으로 이루어진 집합

• 스트림즈는 트리

  

• 토폴로지를 이루는 노드 : processor

• 노드와 노드 이은 선 : stream (topic의 data. record와 동일)

• 소스 프로세서

  • 최초로 데이터를 받는 부분
  • 하나 이상의 topic에서 데이터를 가져오는 역할

• 스트림 프로세서

  • 다른 프로세서가 반환한 데이터를 처리하는 역할
  • 변환, 분기 처리 등

• 싱크 프로세서

  • 데이터를 특정 카프카 토픽으로 저장하는 역할
  • 최종 종착지

스트림즈DSL(Domain Specific Language) 프로세서API 두개 조합하여 개발 가능 스트림즈DSL에서 제공하지 않는 일부 기능 > 프로세서 API로 개발 ex)

레코드의 흐름을 추상화한 3가지 개념 토픽에 있는 데이터를 어떤 형태로 선언할 것인지 KStream, KTable, GlobalKTable

KStream

레코드의 흐름을 표현 메시지 키 + value 데이터 조회하면 토픽에 존재하는(또는 KStream에 존재하는) 모든 레코드가 출력됨 컨슈머로 토픽을 구독하는 것과 동일한 선상

KTable

메시지 키를 기준으로 묶어 사용 유니크한 메시지 키를 기준으로 가장 최신 레코드를 사용

• 가장 최신의 데이터만 유지/활용 (ex. 주소)

co-partitioning

KStream, KTable 데이터를 join join을 하는 2개 데이터의 partition 개수 동일 partitioning strategy를 동일하게 맞추는 작업

파티션 개수 동일 & 파티셔닝 전략이 같다 > 동일한 메시지 키를 가진 데이터가 동일한 task에 들어가는 것 보장 각각의 파티션마다 task에 연결되는 (task마다 파티션의 데이터를 할당받아 사용)

• 조건이 다르면, 같은 태스크에 들어갈 수 없을 수도
• TopologyException

GlobalKTable

co-partitioning되지 않은 KStream과 KTable을 join해서 사용하고 싶을 때 이걸로 정의된 데이터는 스트림즈 애플리케이션의 모든 task에 동일하게 공유되어 사용됨

• materializedview 로 각각의 task에서 활용
• 전체 파티션 모든 데이터를 각각의 task가 가지고 있음
  • 따라서 데이터가 너무 많으면 각 task 가 들고 있는게 너무 많아짐(데이터가 충분히 적을 때 사용하자)
  • 어쩔 수 없이 사용해야 하면 retention 기간을 좀 짧게 두고 사용 (되도록 KTable, KStream 사용)

옵션

필수 옵션

• bootstrap.servers
  • 프로듀서가 데이터를 전송할 대상 카프카 클러스터에 속한 브로커의 호스트이름:포트 1개 이상 작성
• application.id
  • consumer의 group.id와 동일 (컨슈머 그룹)
  • 동일한 id는 동일한 로직 (병렬 처리)
  • 스트림즈 애플리케이션 구분하기 위한 고유 아이디

선택 옵션

• default.key.serde
  • 직/역직렬화 클래스 지정
  • 기본값 : byte (Serdes.ByteArray().getClass().getName()
• default.value.serde
  • 기본값 : byte (Serdes.ByteArray().getClass().getName()
  • KTable<String, Long> 과 같이 쓸 때 Long 형태 사용
• num.stream.threads
  • 스트림 프로세싱 실행 시 실행될 thread 개수
  • 기본값 1
• state.dir
  • stateful 데이터 처리시 데이터를 저장할 directory 지정 (메모리에 올려서 사용하는 게 아닌 파일시스템에 저장해 활용)
  • 디스크에 값 저장
  • 기본값 /tmp/kafka-streams (특정 디렉토리 만들어 쓰자 왠만하면!)
  • RocksDB 활용

필터링 스트림즈 애플리케이션

KTable, KStream join

카프카 파티션 개수만큼 병렬처리 가능 메시지 키를 기준으로 조인 KTable, KStream 조인하기 위해서는 파티션 개수 맞추는게 기본 원칙 실시간으로 들어오는 데이터들을 조인할 수 있음(DB없이 조인, 스트리밍 처리) > 이벤트 기반 스트리밍 데이터 파이프라인 구성

카프카 토픽을 만들 때 KTable, KStream으로 만드는건 아님 스트림즈에서 지정해서 사용

task 개수, 파티션 개수를 늘리면 처리량 늘릴 수 있음. 병렬로

co partitioning 되지 않은 데이터 조인 방법

1. re partitioning > co partitioning된 상태로 join 처리
   • 파티션 개수 2개 였던 거를 다른 애와 맞추기 위해 3개로 늘리거나 하는거
   • 새로운 파티션 추가 + 중복해서 데이터 추가되니까 별로...
2. KTable로 사용하는 토픽을 GlobalKTable로 선언
   • 그리많지 않은 데이터라면 이를 사용하자
   • 모든 globalKTable은 KStream의 모든 파티션과 연결됨 (단점)

window processing

• 특정 시간에 대응하여 취합 연산을 처리할 때 활용
• 메시지 키를 기준으로 취합
• 해당 토픽에 동일한 파티션에는 동일한 메시지 키가 있는 레코드가 존재해야 정확한 취합 가능
• 커스텀 파티셔너를 사용해 동일 메시지 키가 동일 파티션에 저장 보장못하거나, 메시지 키 넣지 않으면 연산 불가능
• 아래 예제들의 노란색/빨간색은 같은 색끼리는 같은 메시지키를 의미

텀블링 윈도우

• 서로 겹치지 않은 윈도우를 특정 간격으로 지속적으로 처리할 때 사용
• 윈도우 최대 사이즈 도달 > 해당 시점에 데이터 취합하여 결과 도출
• 단위 시간당 데이터가 필요할 경우 사용 가능
• 4번만 DB에 insert
• 낮은 비율로 request 보냄
• 매 5분간 접속한 고객의 수 측정 > 방문자 추이를 실시간 취합
• groupByKey, windowwedBy

호핑 윈도우

• 일정 시간 간격으로 윈도우 겹침
• 윈도우 사이즈, 간격 2개의 변수 가짐
• 사이즈 : 연산을 수행할 최대 윈도우 사이즈, 간격 : 서로 다른 윈도우 간 간격
• 동일한 키의 데이터는 서로 다른 윈도우에서 여러번 연산 가능

슬라이딩 윈도우

• 호핑과 유사 but 데이터의 정확한 시간을 바탕으로 윈도우 사이즈에 포함되는 데이터를 모두 연산에 포함시킴

세션 윈도우

• 동일 메시지 키의 데이터를 한 세션에 묶어 연산할 때 사용
• 세션의 최대 만료시간에 따라 윈도우 사이즈가 달라짐
• 세션 만료 시간이 지나게 되면 세션 윈도우 종료 > 해당 윈도우의 모든 데이터를 취합하여 연산
• 세션 윈도우 사이즈는 가변적

카프카 스트림즈는 커밋(기본 값 30초)을 수행할 때 윈도우 사이즈가 종료되지 않아도 중간 정산 데이터를 출력함 커밋 시점마다 윈도우의 연산 데이터를 출력. 동일 윈도우 사이즈(시간)의 데이터는 2개 이상 출력될 수 잇음

• windowed 기준으로 동일 윈도우 시간 데이터를 겹쳐쓰기(upsert) 하는 방식 사용하자!

Queryable store

KTable : 카프카 토픽의 데이터를 로컬의 rocksDB에 Materialized View로 만들어 두고 사용하기 때문에 레코드 메시지키, 값을 기반으로 keyValueStore로 사용할 수 있음 카프카를 사용하여 로컬 캐시를 구현한 것과 유사 ReadOnlyKeyValueSTore로 뷰를 가져오면 메시지 키를 기반으로 토픽 데이터 조회 가능

processor API

스트림즈DSL보다 투박한 코드 but topology를 기준으로 데이터를 처리한다는 관점에서는 동일한 역할 스트림즈DSL은 데이터 처리, 분기, 조인을 다양한 메서드로 제공, 추가적인 상세 로직 구현 필요 > 프로세서 API 활용 프로세서 API는 KStream, KTable, GlobalKTable 개념 없음 스트림즈DSL과 프로세서API는 함께 구현하여 사용할 때는 활용할 수 있음

Interface

• Processor
  • 일정 로직이 이루어진 뒤 다음 프로세서로 데이터가 넘어가지 않을 때 사용
  • 어떨 땐 데이터 넘기고 어떨땐 안넘기고 할 때 이걸 사용할 수 있음
• Transformer
  • 일정 로직이 이루어진 뒤 다음 프로세서로 데이터를 넘길 때 사용

[manjeong-dev]

mirror maker (카프카 클러스터 복제)

• confluent
• num-stream : 스레드 개수

[manjeong-dev]

window processing을 할 때 커밋 간격에 상관없이 윈도우 size에 맞는 데이터를 보려면 upsert 방식을 사용해야 한다

upsert할 수 있는 db Or KTable을 구성해야함

[manjeong-dev]

Q. 프로세서 API 구현 시, Processor 인터페이스를 사용하면 다음 프로세서로 데이터를 넘길 수 없다 (O / X) X (context를 이용해서 넘길 수 있음) Q. 코파티션이 되어 있지 않은 토픽의 데이터를 조인하기 위해서는 GlobalKTable 을 사용하는것이 언제나 좋다. (O/X) X (국가코드 등 개수가 많지 않은 거 사용할 때) Q. KTable과 KStream을 사용하기위해서 토픽생성할때부터 지정해야 한다 (O / X) X (stream 개발시 지정) Q. KStream과 KTable의 Join 메서드를 수행했을 때 메시지키를 수동으로 매칭시켜줘야한다 (O / X) X (join 메서드가 자동으로 매칭해줌) Q. 스트림즈DSL의 텀블링 윈도우를 적용해서 단위시간당 레코드를 중복없이 취합할 수 있다. (O/X) X (commit 시점에 따라 달라질 수 있음) Q. window processing을 할 때 커밋 간격에 상관없이 윈도우 size에 맞는 데이터를 보려면 upsert 방식을 사용해야 한다 (O/X) O Q. 프로세서API의 Processor 인터페이스로 구현한 클래스는 Consumer와 동일한 기능을 한다. X (consumer 파티션 개수에 맞는 스레드 개수 내가 관리, processor 스레드 개수 관리 x worker thread를 내부적인 실행에 맞게 구성해줌) Q. 윈도우 연산시 output은 커밋 시간에 따라 결정되므로 원하는 결과를 얻기 위해서는 윈도우 사이즈(시간)와 커밋 간격을 똑같이 맞추는 것이 좋다. (O/X) O (현실적으로는 완벽하게 맞추기 어려움…) Q. 텀블링 윈도우는 일정 시간 간격으로 겹치는 구간이 존재한다(O/X) X Q. 윈도우 연산 중 데이터의 정확한 시간을 바탕으로 윈도우 사이즈에 포함되는 데이터를 모두 연산에 포함시키는 특징을 가진 것은 ( ) 윈도우이다. 슬라이딩