Kafka Streams

두 개의 API와 애플리케이션의 뼈대

Apache Kafka에는 두 개의 스트림 API가 있다. 저수준 Processor API와 고수준 Streams DSL이다. 예제에서는 Kafka Streams DSL을 쓴다. DSL은 스트림의 이벤트에 적용될 변환의 연쇄를 정의함으로써 스트림 처리 애플리케이션을 정의하게 해 준다. 변환은 filter처럼 단순할 수도, 스트림-스트림 조인처럼 복잡할 수도 있다. 저수준 API는 자기만의 변환을 만들 수 있게 해 주지만, 실제로 필요한 경우는 드물다.

DSL API를 쓰는 애플리케이션은 언제나 StreamBuilder로 처리 토폴로지 — 스트림의 이벤트에 적용되는 변환들의 방향 그래프(DAG) — 를 만드는 데서 시작한다. 그런 다음 그 토폴로지로부터 KafkaStreams 실행 객체를 만든다. KafkaStreams 객체를 시작하면 여러 개의 스레드가 시작되고, 각 스레드가 스트림의 이벤트에 처리 토폴로지를 적용한다. 처리는 KafkaStreams 객체를 닫을 때 종료된다.

Word Count — 설정과 토폴로지

스트림 처리 애플리케이션을 만들 때 가장 먼저 하는 일은 Kafka Streams를 설정하는 것이다. 프로듀서나 컨슈머 설정을 Properties 객체에 추가하면 Kafka Streams에 임베드된 프로듀서·컨슈머도 함께 설정할 수 있다.

public class WordCountExample {
    public static void main(String[] args) throws Exception{
        Properties props = new Properties();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "wordcount");
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(StreamsConfig.KEY_SERDE_CLASS_CONFIG,
          Serdes.String().getClass().getName());
        props.put(StreamsConfig.VALUE_SERDE_CLASS_CONFIG,
          Serdes.String().getClass().getName());

세 설정의 의미는 이렇다. 모든 Kafka Streams 애플리케이션은 애플리케이션 ID를 가져야 한다. 이 값은 애플리케이션 인스턴스들을 조정하는 데 쓰이고, 내부 로컬 저장소와 그에 관련된 토픽의 이름을 짓는 데도 쓰인다. 같은 Kafka 클러스터를 쓰는 Kafka Streams 애플리케이션마다 이 이름은 유일해야 한다. bootstrap.servers가 필요한 이유는 Kafka Streams 애플리케이션이 언제나 Kafka 토픽에서 데이터를 읽고 Kafka 토픽에 출력을 쓰며, 조정에도 Kafka를 쓰기 때문이다. 마지막으로 데이터를 읽고 쓸 때 직렬화·역직렬화가 필요하므로 기본 Serde 클래스를 제공한다. 필요하면 토폴로지를 만들 때 이 기본값을 덮어쓸 수 있다.

토폴로지는 이렇게 만든다.

KStreamBuilder builder = new KStreamBuilder();
KStream<String, String> source = builder.stream("wordcount-input");
final Pattern pattern = Pattern.compile("\\W+");
KStream counts = source.flatMapValues(value->
      Arrays.asList(pattern.split(value.toLowerCase())))
        .map((key, value) -> new KeyValue<Object, Object>(value, value))
        .filter((key, value) -> (!value.equals("the")))
        .groupByKey()
        .count("CountStore").mapValues(value->
          Long.toString(value)).toStream();
counts.to("wordcount-output");

흐름은 이렇다. KStreamBuilder를 만들고 입력으로 쓸 토픽을 가리켜 스트림 정의를 시작한다. 소스 토픽에서 읽는 각 이벤트는 단어들로 이루어진 한 줄이므로, 정규식으로 개별 단어들로 쪼갠다. 그런 다음 각 단어(현재는 이벤트 레코드의 값)를 이벤트 레코드의 키에 넣어 group-by 연산에 쓸 수 있게 한다. 필터링이 얼마나 쉬운지 보여 주기 위해 'the'라는 단어를 걸러 낸다. 키로 그룹핑하면 각 고유 단어마다 이벤트의 모음을 갖게 된다. 각 모음의 이벤트 개수를 센다. 세기의 결과는 Long 타입이므로 사람이 읽기 쉽도록 String으로 변환한다. 마지막으로 결과를 Kafka에 되돌려 쓴다.

실행은 다음과 같다.

KafkaStreams streams = new KafkaStreams(builder, props);
streams.start();
// 보통 스트림 애플리케이션은 영원히 돌지만,
// 이 예제는 입력 데이터가 유한하므로 잠시 돌린 뒤 멈춘다.
Thread.sleep(5000L);
streams.close();

이 짧은 코드가 단일 이벤트 처리 패턴(map과 filter 적용), group-by 연산자를 더한 리파티셔닝, 그리고 각 단어를 키로 레코드 수를 세는 단순 로컬 상태 유지를 모두 보여 준다.

주목할 점 하나는 Apache Kafka 외에는 아무것도 설치하지 않고 이 예제를 자기 컴퓨터에서 통째로 돌릴 수 있다는 것이다. Spark의 로컬 모드와 비슷한 경험이다. 다만 결정적인 차이가 있다. 입력 토픽에 파티션이 여러 개라면 WordCount 애플리케이션 인스턴스를 여러 개 실행하는 것만으로(여러 터미널 탭에서 그냥 실행하면 된다) 첫 Kafka Streams 처리 클러스터가 된다. 인스턴스들끼리 대화하며 작업을 조정한다. Spark의 진입 장벽 중 하나는 로컬 모드는 아주 쉽지만 프로덕션 클러스터를 돌리려면 YARN이나 Mesos를 설치하고 모든 머신에 Spark를 설치한 뒤 애플리케이션 제출 방법을 배워야 한다는 점이다. Kafka의 Streams API에서는 애플리케이션 인스턴스를 여러 개 시작하기만 하면 그것이 클러스터이며, 개발 머신과 프로덕션에서 정확히 같은 애플리케이션이 돈다.

주식 시장 통계 — 윈도우 집계와 상태 저장소

두 번째 예제는 더 복잡하다. 종목 기호(ticker), 호가(ask price), 호가 수량(ask size)이 담긴 주식 거래 이벤트 스트림을 읽는다. 주식 거래에서 ask price는 파는 쪽이 요구하는 가격이고 bid price는 사는 쪽이 제시하는 가격이며, ask size는 그 가격에 팔려는 주식 수다. 예제를 단순하게 하려고 bid는 완전히 무시한다. 데이터에 타임스탬프도 넣지 않고, 대신 Kafka 프로듀서가 채워 주는 event time에 의존한다.

만들 출력 스트림은 다음 세 가지 윈도우 통계이며, 모두 1초마다 갱신된다.

- 5초 윈도우마다의 최적(=최소) 호가 - 5초 윈도우마다의 거래 건수 - 5초 윈도우마다의 평균 호가

설정은 Word Count와 거의 같고, 주된 차이는 Serde 클래스다. Word Count에서는 키와 값 모두 문자열이라 Serdes.String()을 썼지만, 여기서는 키는 여전히 문자열이되 값이 종목 기호·호가·호가 수량을 담은 Trade 객체다. 이 객체를 직렬화·역직렬화하기 위해 Google의 Gson 라이브러리로 Java 객체에서 JSON 직렬화기·역직렬화기를 생성하고, 그것들로부터 Serde 객체를 만드는 작은 래퍼를 만들었다.

static public final class TradeSerde extends WrapperSerde<Trade> {
    public TradeSerde() {
        super(new JsonSerializer<Trade>(),
          new JsonDeserializer<Trade>(Trade.class));
    }
}

대단한 것은 없지만, Kafka에 저장하려는 모든 객체 — 입력, 출력, 경우에 따라 중간 결과까지 — 에 대해 Serde 객체를 제공해야 한다는 점은 기억해야 한다. 이를 쉽게 하려면 GSon, Avro, Protobufs 같은 프로젝트로 Serde를 생성하는 것이 권장된다.

토폴로지는 다음과 같다.

KStream<TickerWindow, TradeStats> stats = source.groupByKey()
        .aggregate(TradeStats::new,
            (k, v, tradestats) -> tradestats.add(v),
            TimeWindows.of(5000).advanceBy(1000),
            new TradeStatsSerde(),
            "trade-stats-store")
        .toStream((key, value) -> new TickerWindow(key.key(),
          key.window().start()))
        .mapValues((trade) -> trade.computeAvgPrice());
stats.to(new TickerWindowSerde(), new TradeStatsSerde(),
  "stockstats-output");

각 단계의 의미는 이렇다.

먼저 입력 토픽에서 이벤트를 읽고 groupByKey()를 수행한다. **이름과 달리 이 연산은 그룹핑을 하지 않는다.** 오히려 이벤트 스트림이 레코드 키를 기준으로 파티셔닝되도록 보장하는 역할이다. 여기서는 키를 가진 채로 토픽에 데이터를 썼고 groupByKey() 호출 전에 키를 바꾸지 않았으므로, 데이터는 여전히 키로 파티셔닝되어 있고 따라서 이 메서드는 아무 일도 하지 않는다.

올바른 파티셔닝이 보장된 뒤에 윈도우 집계를 시작한다. aggregate 메서드는 스트림을 겹치는 윈도우들(1초마다 만들어지는 5초 윈도우)로 나눈 다음, 윈도우 안의 모든 이벤트에 집계 메서드를 적용한다. 첫 번째 파라미터는 집계 결과를 담을 새 객체 — 여기서는 TradeStats — 다. 이 객체는 각 시간 윈도우에 대해 우리가 관심 있는 통계, 즉 최저가·평균가·거래 건수를 담기 위해 만든 것이다. 두 번째로 실제로 레코드를 집계하는 메서드를 제공한다. 여기서는 TradeStats의 add 메서드가 새 레코드로 윈도우 안의 최저가·거래 건수·가격 합계를 갱신한다. 세 번째로 윈도우를 정의한다 — 5초(5,000ms) 윈도우가 1초마다 전진한다. 네 번째로 집계 결과(TradeStats 객체)를 직렬화·역직렬화할 Serde 객체를 제공한다. 마지막 파라미터는 상태 저장소의 이름이다. 윈도우 집계는 상태와 그것을 유지할 로컬 저장소를 필요로 하며, 이 이름은 유일하기만 하면 무엇이든 된다.

집계의 결과는 종목 기호와 시간 윈도우를 기본 키로, 집계 결과를 값으로 갖는 테이블이다. toStream 메서드로 이 테이블을 다시 이벤트 스트림으로 바꾸면서, 시간 윈도우 정의 전체를 담고 있던 키를 종목 기호와 윈도우 시작 시각만 담은 TickerWindow 객체로 교체한다. 마지막으로 평균가를 갱신한다. 이 시점의 집계 결과에는 가격 합계와 거래 건수가 들어 있으므로, 이 값들로 평균가를 계산해 출력 스트림에 포함시킨다. 그리고 결과를 stockstats-output 스트림에 쓴다.

이 예제는 스트림에 대한 윈도우 집계 — 아마도 스트림 처리의 가장 인기 있는 사용 사례 — 를 보여 준다. 눈여겨볼 점은 집계의 로컬 상태를 유지하는 데 필요한 작업이 Serde 제공과 상태 저장소 이름 붙이기뿐이라는 것이다. 그런데도 이 애플리케이션은 여러 인스턴스로 확장되고, 인스턴스 하나가 죽으면 그 파티션의 처리를 살아남은 인스턴스로 옮겨 자동으로 복구한다.

클릭스트림 보강 — 두 종류의 조인

마지막 예제는 웹사이트의 클릭 스트림을 보강해 스트리밍 조인을 보여 준다. 시뮬레이션된 클릭 스트림, 가상의 프로필 데이터베이스 테이블에 대한 갱신 스트림, 웹 검색 스트림을 만들어 셋을 조인하고 사용자 활동에 대한 360도 관점을 얻는다. 사용자가 무엇을 검색했는가? 그 결과로 무엇을 클릭했는가? 프로필의 '관심사'를 바꿨는가? 이런 조인은 분석을 위한 풍부한 데이터를 만들어 낸다. 상품 추천이 흔히 이런 정보에 기반한다 — 사용자가 자전거를 검색하고 'Trek' 링크를 클릭했으며 여행에 관심이 있으니, Trek 자전거와 헬멧, 자전거 여행 상품을 광고할 수 있다.

KStream<Integer, PageView> views =
  builder.stream(Serdes.Integer(), new PageViewSerde(),
    Constants.PAGE_VIEW_TOPIC);
KStream<Integer, Search> searches =
  builder.stream(Serdes.Integer(), new SearchSerde(),
    Constants.SEARCH_TOPIC);
KTable<Integer, UserProfile> profiles =
  builder.table(Serdes.Integer(), new ProfileSerde(),
    Constants.USER_PROFILE_TOPIC, "profile-store");

KStream<Integer, UserActivity> viewsWithProfile = views.leftJoin(profiles,
    (page, profile) -> new UserActivity(profile.getUserID(),
    profile.getUserName(), profile.getZipcode(),
    profile.getInterests(), "", page.getPage()));

KStream<Integer, UserActivity> userActivityKStream =
  viewsWithProfile.leftJoin(searches,
    (userActivity, search) ->
    userActivity.updateSearch(search.getSearchTerms()),
    JoinWindows.of(1000), Serdes.Integer(),
    new UserActivitySerde(), new SearchSerde());

먼저 조인할 두 스트림(클릭과 검색)에 대한 스트림 객체를 만든다. 사용자 프로필에 대해서는 KTable을 정의한다. KTable은 변경 스트림을 통해 갱신되는 로컬 캐시다.

그런 다음 이벤트 스트림을 프로필 테이블과 조인해 클릭 스트림을 사용자 프로필 정보로 보강한다. 스트림-테이블 조인에서는 스트림의 각 이벤트가 캐싱된 프로필 테이블 사본으로부터 정보를 받는다. 여기서는 left join이므로 알려지지 않은 사용자의 클릭도 보존된다. 조인 메서드는 스트림에서 하나, 레코드에서 하나, 두 값을 받아 세 번째 값을 반환한다. 데이터베이스와 달리 두 값을 어떻게 하나의 결과로 합칠지는 우리가 정한다. 여기서는 사용자 상세와 조회한 페이지를 모두 담은 활동 객체 하나를 만들었다.

다음으로 같은 사용자가 수행한 검색과 클릭 정보를 조인한다. 여전히 left join이지만 이번에는 스트림과 테이블이 아니라 두 개의 스트림을 조인한다. 조인 메서드는 매칭되는 모든 페이지 뷰에 검색어를 더하기만 한다.

흥미로운 부분은 이것이다. **스트림-스트림 조인은 시간 윈도우가 있는 조인이다.** 사용자별로 모든 클릭과 모든 검색을 조인하는 것은 별 의미가 없다. 우리가 원하는 것은 각 검색을 그와 관련된 클릭, 즉 검색 직후 짧은 시간 안에 일어난 클릭과 맞추는 것이다. 그래서 1초의 조인 윈도우를 정의한다. 검색으로부터 1초 안에 일어난 클릭이 관련 있는 것으로 간주되고, 검색어가 클릭과 사용자 프로필을 담은 활동 레코드에 포함된다. 이로써 검색과 그 결과에 대한 완전한 분석이 가능해진다.

이 예제는 스트림 처리에서 가능한 두 가지 조인 패턴을 보여 준다. 하나는 스트림을 테이블과 조인해 모든 스트리밍 이벤트를 테이블의 정보로 보강하는 것으로, 데이터 웨어하우스에서 팩트 테이블을 디멘션과 조인하는 것과 비슷하다. 다른 하나는 시간 윈도우를 기준으로 두 스트림을 조인하는 것이며, 이 연산은 스트림 처리에 고유하다.