카프카가 제공하는 스트림 처리 라이브러리. 별도의 클러스터가 아니라 라이브러리라는 점이 이 도구의 성격을 결정한다.
처리 토폴로지
모든 스트림즈 애플리케이션은 하나의 처리 토폴로지를 구현하고 실행한다. 스트림 안의 이벤트에 적용되는 변환을 정점으로 하는 유향 비순환 그래프(DAG)이고, 토픽에서 읽어 넘겨주는 소스 프로세서로 시작해서 필터와 맵과 집계 같은 처리를 하는 프로세서들을 거쳐 토픽에 쓰는 싱크 프로세서로 끝난다.
Processor API와 스트림 DSL
API는 저수준과 고수준으로 나뉜다. 저수준 Processor API는 변환을 직접 만들게 해주고, 고수준 스트림 DSL은 연속적인 변환을 선언하게 해준다. DSL 애플리케이션은 항상 StreamsBuilder로 처리 토폴로지를 만들고 거기서 KafkaStreams 실행 객체를 만드는 것으로 시작한다.
StreamsBuilder builder = new StreamsBuilder();
KStream<String, String> source = builder.stream("wordcount-input");
KStream counts = source
.flatMapValues(value -> Arrays.asList(pattern.split(value.toLowerCase())))
.map((key, value) -> new KeyValue<>(value, value))
.filter((key, value) -> !value.equals("the"))
.groupByKey()
.count()
.toStream();
counts.to("wordcount-output");단어를 이벤트 레코드의 키로 넣어주는 부분이 없으면 groupByKey()로 묶을 수가 없다. 키가 파티션을 정하고, 파티션이 상태의 범위를 정한다는 원리가 여기서도 그대로다.
필수 설정
설정은 세 가지가 필수다. APPLICATION_ID_CONFIG는 인스턴스들이 협력하는 데 쓰이고 내부 로컬 저장소와 연관 토픽의 이름을 정하는 데도 쓰이므로, 같은 클러스터를 쓰는 애플리케이션마다 달라야 한다. BOOTSTRAP_SERVERS_CONFIG는 토픽을 읽고 쓸 뿐 아니라 인스턴스끼리 협력하는 데도 카프카를 쓰기 때문에 필요하다. DEFAULT_KEY_SERDE_CLASS_CONFIG와 DEFAULT_VALUE_SERDE_CLASS_CONFIG는 직렬화와 역직렬화에 쓸 Serde 클래스를 정하고 토폴로지를 만들 때 재정의할 수 있다. 키와 값이 둘 다 문자열이면 Serdes.String()이면 되고, 다른 객체라면 Gson이나 Avro, Protobuf로 Serde를 만드는 것이 권장된다.
라이브러리라서 단순해지는 배포
배포에서 다른 프레임워크와 결정적으로 갈린다. 보통은 YARN이나 Mesos를 설치하고 처리 프레임워크를 모든 장비에 설치하고 애플리케이션을 클러스터에 제출해야 한다. 스트림즈는 애플리케이션 인스턴스를 여러 개 띄우는 것만으로 처리 클러스터가 된다. 라이브러리이기 때문이다.
태스크 단위의 병렬 처리
확장도 같은 이유로 단순하다. 하나의 인스턴스 안에 여러 스레드를 띄울 수 있고 여러 장비에서 실행할 수도 있는데, 어느 쪽이든 모든 활성 스레드가 작업을 균등하게 나눈다. 병렬 처리의 기본 단위는 태스크다. 스트림즈 엔진이 처리하는 토픽의 파티션 수에 따라 태스크 수를 결정하고, 각 태스크가 담당 파티션을 구독해 이벤트를 읽고 그 파티션에 적용될 모든 처리 단계를 실행한 뒤 싱크에 쓴다. 태스크들은 서로 완전히 독립적으로 실행된다. 그래서 더 빨리 처리하려면 스레드를 늘리고, 서버 자원이 고갈되면 다른 서버에 인스턴스를 추가하면 된다.
리파티셔닝과 서브 토폴로지
리파티셔닝이 필요할 때 다른 프레임워크는 셔플을 하거나 다른 태스크로 이벤트를 보내는데, 스트림즈는 새로운 키와 파티션으로 새 토픽에 이벤트를 쓰고 다음 태스크들이 그 토픽에서 읽어간다. 이렇게 하면 전체 토폴로지가 두 개의 서브 토폴로지로 분할된다. 두 번째가 첫 번째의 결과를 받지만, 첫 번째는 자기 속도로 토픽에 쓰고 두 번째는 자기 속도로 읽으므로 둘은 독립적이고 병렬로 실행된다. 태스크 사이에 통신도 공유 리소스도 없으니 같은 스레드나 서버에 있을 필요조차 없다. 카프카를 중간에 두는 것의 이점이 여기서 드러난다.
세 겹의 장애 처리
장애 처리는 세 겹이다. 재시작하면 마지막으로 커밋된 오프셋을 카프카에서 가져와 처리하던 지점부터 재개하고, 로컬 상태 저장소가 유실되면 체인지 로그를 읽어 복구하고, 태스크에 장애가 나면 사용 가능한 다른 스레드에서 재시작한다. 셋째는 컨슈머 그룹에서 컨슈머 하나가 죽었을 때 그 파티션을 남은 컨슈머에게 할당하는 것과 정확히 같은 메커니즘이라, 정적 그룹 멤버십이나 협력적 리밸런스 같은 컨슈머 쪽 개선이 그대로 스트림즈의 개선이 된다.
정확히 한 번 보장
정확히 한 번 보장은 processing.guarantee를 exactly_once_v2로 잡으면 켜진다. 카프카 트랜잭션을 스트림즈가 대신 써준다.
조인이 만드는 태스크 의존
조인은 태스크 사이에 의존 관계를 만든다. 스트림즈는 조인에 필요한 모든 파티션을 하나의 태스크에 할당해서 그 태스크가 필요한 데이터를 전부 볼 수 있게 하는데, 그러려면 조인에 쓰이는 모든 토픽이 동일한 조인 키로 파티션되어 있고 파티션 수도 같아야 한다.
기본으로 꺼져 있는 최적화
최적화는 기본으로 꺼져 있다. DSL의 각 메서드가 독립적으로 저수준 API로 변환되므로 결과 토폴로지가 전체적으로는 최적화되지 않은 상태일 수 있다. TOPOLOGY_OPTIMIZATION을 OPTIMIZE로 잡고 build(props)를 호출해야 켜지고, 설정 없이 build()를 호출하면 적용되지 않는다. 테스트할 때 적용한 것과 안 한 것의 실행 시간, 카프카에 쓰이는 데이터 양을 비교해보는 것이 좋다.
테스트 도구의 구멍
테스트 도구에도 구멍이 있다. TopologyTestDriver가 주된 도구라서 입력 데이터를 정의하고 목업 입력 토픽에 쓰고 토폴로지를 실행시키고 목업 출력 토픽에서 결과를 읽어 검증하는데, 스트림즈의 캐시 기능을 시뮬레이션하지 않아서 놓치는 에러가 많다. 그래서 통합 테스트로 보강한다. EmbeddedKafkaCluster는 테스트 JVM 위에 브로커를 하나 띄우고, Testcontainers는 도커 컨테이너로 브로커와 필요한 요소를 띄운다. 카프카와 그 의존성을 테스트 애플리케이션으로부터 완전히 격리하므로 Testcontainers가 더 권장된다.