"Kafka Streams in Action"이라는 책이 있습니다. 실시간 작업을 위한 애플리케이션 및 마이크로서비스"

안녕하세요, 카브로 주민 여러분! 이 책은 스레드 처리를 이해하려는 모든 개발자에게 적합합니다. 분산 프로그래밍을 이해하면 Kafka와 Kafka Streams를 더 잘 이해하는 데 도움이 됩니다. Kafka 프레임워크 자체를 알면 좋겠지만 꼭 필요한 것은 아닙니다. 필요한 모든 것을 알려드리겠습니다. 숙련된 Kafka 개발자와 초보자 모두 이 책에서 Kafka Streams 라이브러리를 사용하여 흥미로운 스트림 처리 애플리케이션을 만드는 방법을 배울 것입니다. 직렬화와 같은 개념에 이미 익숙한 중급 및 고급 Java 개발자는 자신의 기술을 적용하여 Kafka Streams 애플리케이션을 만드는 방법을 배웁니다. 이 책의 소스 코드는 Java 8로 작성되었으며 Java 8 람다 표현식 구문을 많이 사용하므로 (다른 프로그래밍 언어에서도) 람다 함수를 사용하는 방법을 아는 것이 도움이 될 것입니다.

발췌. 5.3. 집계 및 창 작업

이 섹션에서는 Kafka Streams의 가장 유망한 부분을 살펴보겠습니다. 지금까지 Kafka Streams의 다음 측면을 다루었습니다.

• 처리 토폴로지를 생성하고;
• 스트리밍 애플리케이션에서 상태 사용;
• 데이터 스트림 연결을 수행하고;
• 이벤트 스트림(KStream)과 업데이트 스트림(KTable)의 차이점

다음 예에서는 이러한 모든 요소를 ​​하나로 모을 것입니다. 스트리밍 애플리케이션의 또 다른 뛰어난 기능인 윈도우잉에 대해서도 알아봅니다. 첫 번째 예는 간단한 집계입니다.

5.3.1. 업종별 주식매출 집계

집계 및 그룹화는 스트리밍 데이터 작업 시 필수적인 도구입니다. 개인 기록을 접수하면서 검토하는 것만으로는 충분하지 않은 경우가 많습니다. 데이터에서 추가 정보를 추출하려면 데이터를 그룹화하고 결합해야 합니다.

이 예에서는 여러 산업 분야의 회사 주식 판매량을 추적해야 하는 일일 거래자의 의상을 입게 됩니다. 특히, 귀하는 각 산업 분야에서 주식 판매량이 가장 많은 XNUMX개 회사에 관심이 있습니다.

이러한 집계에는 데이터를 원하는 형식(일반적인 용어로 말하면)으로 변환하기 위해 다음과 같은 여러 단계가 필요합니다.

1. 원시 주식 거래 정보를 게시하는 주제 기반 소스를 만듭니다. StockTransaction 유형의 개체를 ShareVolume 유형의 개체에 매핑해야 합니다. 요점은 StockTransaction 개체에 판매 메타데이터가 포함되어 있지만 판매되는 주식 수에 대한 데이터만 필요하다는 것입니다.
2. 주식 기호별로 ShareVolume 데이터를 그룹화합니다. 기호별로 그룹화한 후에는 이 데이터를 주식 판매량의 소계로 축소할 수 있습니다. KStream.groupBy 메소드가 KGroupedStream 유형의 인스턴스를 반환한다는 점은 주목할 가치가 있습니다. 그리고 KGroupedStream.reduce 메소드를 추가로 호출하여 KTable 인스턴스를 얻을 수 있습니다.

KGroupedStream 인터페이스는 무엇입니까

KStream.groupBy 및 KStream.groupByKey 메서드는 KGroupedStream의 인스턴스를 반환합니다. KGroupedStream은 키별로 그룹화한 후 이벤트 스트림의 중간 표현입니다. 직접적인 작업을 위한 것이 전혀 아닙니다. 대신, KGroupedStream은 항상 KTable을 생성하는 집계 작업에 사용됩니다. 그리고 집계 작업의 결과는 KTable이고 상태 저장소를 사용하므로 결과적으로 모든 업데이트가 파이프라인 아래로 전송되지 않을 수도 있습니다.

KTable.groupBy 메소드는 유사한 KGroupedTable(키별로 재그룹화된 업데이트 스트림의 중간 표현)을 반환합니다.

잠시 쉬면서 그림을 살펴보겠습니다. 5.9는 우리가 달성한 것을 보여줍니다. 이 토폴로지는 이미 여러분에게 매우 익숙할 것입니다.

이제 이 토폴로지의 코드를 살펴보겠습니다(src/main/java/bbejeck/chapter_5/AggregationsAndReducingExample.java 파일에서 찾을 수 있음)(목록 5.2).

주어진 코드는 간결성과 여러 줄에서 수행되는 많은 양의 작업으로 구별됩니다. builder.stream 메소드의 첫 번째 매개변수에서 새로운 내용을 발견할 수 있습니다. 즉, Consumed.withOffsetResetPolicy 메소드를 사용하여 설정된 열거형 AutoOffsetReset.EARLIEST(LATEST도 있음)의 값입니다. 이 열거 유형은 각 KStream 또는 KTable에 대한 오프셋 재설정 전략을 지정하는 데 사용될 수 있으며 구성의 오프셋 재설정 옵션보다 우선합니다.

GroupByKey 및 GroupBy

KStream 인터페이스에는 레코드를 그룹화하는 두 가지 방법(GroupByKey 및 GroupBy)이 있습니다. 둘 다 KGroupedTable을 반환하므로 둘 사이의 차이점이 무엇인지, 언제 어느 것을 사용해야 하는지 궁금할 것입니다.

GroupByKey 메서드는 KStream의 키가 이미 비어 있지 않을 때 사용됩니다. 그리고 가장 중요한 것은 "재파티셔닝 필요" 플래그가 설정되지 않았다는 것입니다.

GroupBy 메서드는 그룹화 키를 변경했다고 가정하므로 다시 분할 플래그가 true로 설정됩니다. GroupBy 메서드 이후에 조인, 집계 등을 수행하면 자동으로 다시 파티셔닝됩니다.
요약: 가능하면 GroupBy 대신 GroupByKey를 사용해야 합니다.

mapValues ​​및 groupBy 메소드가 수행하는 작업이 무엇인지 명확하므로 sum() 메소드(src/main/java/bbejeck/model/ShareVolume.java에 있음)를 살펴보겠습니다(목록 5.3).

ShareVolume.sum 메소드는 주식 판매량의 누계를 반환하고 전체 계산 체인의 결과는 KTable 개체입니다. . 이제 KTable의 역할을 이해하셨습니다. ShareVolume 개체가 도착하면 해당 KTable 개체는 최신 현재 업데이트를 저장합니다. 모든 업데이트가 이전 shareVolumeKTable에 반영되지만 모든 업데이트가 더 이상 전송되는 것은 아니라는 점을 기억하는 것이 중요합니다.

그런 다음 이 KTable을 사용하여 (거래된 주식 수 기준) 집계하여 각 산업에서 가장 많은 주식이 거래된 XNUMX개 회사에 도달합니다. 이 경우의 조치는 첫 번째 집계의 조치와 ​​유사합니다.

1. 산업별로 개별 ShareVolume 개체를 그룹화하려면 또 다른 groupBy 작업을 수행합니다.
2. ShareVolume 개체 요약을 시작합니다. 이번에는 집계 개체가 고정 크기 우선 순위 대기열입니다. 이 고정 크기 대기열에는 가장 많은 양의 주식을 판매한 XNUMX개 회사만 유지됩니다.
3. 이전 단락의 대기열을 문자열 값으로 매핑하고 업종별로 가장 많이 거래된 상위 XNUMX개 주식을 반환합니다.
4. 결과를 문자열 형식으로 주제에 씁니다.

그림에서. 그림 5.10은 데이터 흐름 토폴로지 그래프를 보여줍니다. 보시다시피 두 번째 처리 라운드는 매우 간단합니다.

이제 두 번째 처리 라운드의 구조를 명확하게 이해했으므로 해당 소스 코드를 살펴볼 수 있습니다(src/main/java/bbejeck/chapter_5/AggregationsAndReducingExample.java 파일에서 찾을 수 있음)(목록 5.4) .

이 초기화 프로그램에는fixedQueue 변수가 포함되어 있습니다. 이는 거래된 주식의 내림차순으로 상위 N개 결과를 추적하는 데 사용되는 java.util.TreeSet용 어댑터인 사용자 정의 개체입니다.

groupBy 및 mapValues ​​호출을 이미 보았으므로 이에 대해서는 다루지 않겠습니다(KTable.print 메서드는 더 이상 사용되지 않으므로 KTable.toStream 메서드를 호출합니다). 하지만 여러분은 아직 KTable 버전의 Aggregate()를 본 적이 없으므로 이에 대해 잠시 논의하겠습니다.

기억하시겠지만, KTable이 다른 점은 동일한 키를 가진 레코드가 업데이트로 간주된다는 것입니다. KTable은 이전 항목을 새 항목으로 대체합니다. 집계는 비슷한 방식으로 발생합니다. 동일한 키를 가진 최신 레코드가 집계됩니다. 레코드가 도착하면 덧셈기(집계 메서드 호출의 두 번째 매개 변수)를 사용하여 FixedSizePriorityQueue 클래스 인스턴스에 추가되지만, 동일한 키를 가진 다른 레코드가 이미 존재하는 경우 빼기(집계 메서드 호출의 세 번째 매개 변수)를 사용하여 이전 레코드가 제거됩니다. 집계 메소드 호출).

이는 우리의 집계자(FixedSizePriorityQueue)가 하나의 키로 모든 값을 집계하는 것이 아니라 가장 많이 거래되는 N개 유형의 주식 수량의 이동 합계를 저장한다는 것을 의미합니다. 각 수신 항목에는 지금까지 판매된 총 주식 수가 포함됩니다. KTable은 각 업데이트의 롤링 집계 없이 현재 가장 많이 거래되는 회사의 주식에 대한 정보를 제공합니다.

우리는 두 가지 중요한 일을 하는 방법을 배웠습니다.

• KTable의 값을 공통 키로 그룹화합니다.
• 이러한 그룹화된 값에 대해 롤업 및 집계와 같은 유용한 작업을 수행합니다.

Kafka Streams 애플리케이션을 통해 이동하는 데이터의 의미를 이해하고 해당 데이터가 전달하는 정보를 이해하려면 이러한 작업을 수행하는 방법을 아는 것이 중요합니다.

또한 이 책의 앞부분에서 논의한 몇 가지 주요 개념을 정리했습니다. 4장에서는 내결함성, 로컬 상태가 스트리밍 애플리케이션에 얼마나 중요한지 논의했습니다. 이 장의 첫 번째 예에서는 지역 상태가 왜 그렇게 중요한지 보여주었습니다. 이는 이미 본 정보를 추적할 수 있는 기능을 제공합니다. 로컬 액세스는 네트워크 지연을 방지하여 애플리케이션의 성능을 높이고 오류 방지 기능을 향상시킵니다.

롤업 또는 집계 작업을 수행할 때 상태 저장소의 이름을 지정해야 합니다. 롤업 및 집계 작업은 KTable 인스턴스를 반환하고 KTable은 상태 저장소를 사용하여 이전 결과를 새 결과로 바꿉니다. 본 것처럼 모든 업데이트가 파이프라인으로 전송되는 것은 아니며 집계 작업은 요약 정보를 생성하도록 설계되었기 때문에 이는 중요합니다. 로컬 상태를 적용하지 않으면 KTable은 모든 집계 및 롤업 결과를 전달합니다.

다음으로 특정 기간 내에 집계와 같은 작업(소위 윈도잉 작업이라고 함)을 수행하는 방법을 살펴보겠습니다.

5.3.2. 창 작업

이전 섹션에서는 슬라이딩 컨볼루션과 집계를 소개했습니다. 애플리케이션은 주식 판매량을 지속적으로 롤업한 후 거래소에서 가장 많이 거래되는 XNUMX개 주식을 집계했습니다.

때로는 이러한 결과의 지속적인 집계 및 롤업이 필요합니다. 때로는 특정 기간 동안에만 작업을 수행해야 하는 경우도 있습니다. 예를 들어, 지난 10분 동안 특정 회사의 주식에 대해 얼마나 많은 교환 거래가 이루어졌는지 계산해 보세요. 또는 지난 15분 동안 새 광고 배너를 클릭한 사용자 수입니다. 애플리케이션은 이러한 작업을 여러 번 수행할 수 있지만 결과는 지정된 기간(기간)에만 적용됩니다.

구매자별 교환거래 집계

다음 예에서는 대규모 조직이든 현명한 개인 금융가든 여러 거래자 간의 주식 거래를 추적합니다.

이 추적에는 두 가지 가능한 이유가 있습니다. 그 중 하나는 시장 리더가 무엇을 사고 파는지 알아야 한다는 것입니다. 이러한 대형 기업과 정교한 투자자가 기회를 발견한다면 그들의 전략을 따르는 것이 합리적입니다. 두 번째 이유는 불법적인 내부 거래의 징후를 찾아내고자 하는 욕구입니다. 이를 위해서는 대규모 매출 급증과 중요한 보도 자료의 상관관계를 분석해야 합니다.

이러한 추적은 다음 단계로 구성됩니다.

• 주식 거래 주제에서 읽기 위한 스트림을 생성합니다.
• 구매자 ID 및 주식 기호별로 수신 레코드를 그룹화합니다. groupBy 메소드를 호출하면 KGroupedStream 클래스의 인스턴스가 반환됩니다.
• KGroupedStream.windowedBy 메서드는 기간 집계를 허용하는 시간 창으로 제한된 데이터 스트림을 반환합니다. 창 유형에 따라 TimeWindowedKStream 또는 SessionWindowedKStream이 반환됩니다.
• 집계 작업에 대한 트랜잭션 수입니다. 기간이 지정된 데이터 흐름은 이 계산에서 특정 레코드가 고려되는지 여부를 결정합니다.
• 결과를 주제에 쓰거나 개발 중에 콘솔에 출력합니다.

이 애플리케이션의 토폴로지는 간단하지만 이에 대한 명확한 그림이 도움이 될 것입니다. 그림을 살펴보자. 5.11.

다음으로 창 작업의 기능과 해당 코드를 살펴보겠습니다.

창 유형

Kafka Streams에는 세 가지 유형의 창이 있습니다.

• 세션별;
• "텀블링";
• 슬라이딩/호핑.

어떤 것을 선택할지는 비즈니스 요구 사항에 따라 다릅니다. 텀블링 및 점프 창은 시간이 제한되어 있는 반면, 세션 창은 사용자 활동에 따라 제한됩니다. 세션 기간은 사용자의 활동 정도에 의해서만 결정됩니다. 기억해야 할 가장 중요한 점은 모든 창 유형은 시스템 시간이 아닌 항목의 날짜/시간 스탬프를 기반으로 한다는 것입니다.

다음으로 각 창 유형으로 토폴로지를 구현합니다. 전체 코드는 첫 번째 예에서만 제공되며 다른 유형의 창에서는 창 작업 유형 외에는 아무것도 변경되지 않습니다.

세션 창

세션 창은 다른 모든 유형의 창과 매우 다릅니다. 이는 시간보다는 사용자의 활동(또는 추적하려는 엔터티의 활동)에 따라 제한됩니다. 세션 기간은 비활성 기간으로 구분됩니다.

그림 5.12는 세션 창의 개념을 보여줍니다. 더 작은 세션이 왼쪽 세션과 병합됩니다. 그리고 오른쪽 세션은 오랜 기간 동안 활동이 없기 때문에 분리됩니다. 세션 창은 사용자 활동을 기반으로 하지만 항목의 날짜/시간 스탬프를 사용하여 항목이 속한 세션을 결정합니다.

세션 창을 사용하여 주식 거래 추적

세션 창을 사용하여 교환 거래에 대한 정보를 캡처해 보겠습니다. 세션 창의 구현은 목록 5.5(src/main/java/bbejeck/chapter_5/CountingWindowingAndKTableJoinExample.java에서 찾을 수 있음)에 나와 있습니다.

이 토폴로지에서 대부분의 작업을 이미 살펴보았으므로 여기서 다시 볼 필요가 없습니다. 그러나 여기에는 몇 가지 새로운 요소도 있는데, 이제 이에 대해 논의하겠습니다.

모든 groupBy 작업은 일반적으로 일종의 집계 작업(집계, 롤업 또는 계산)을 수행합니다. 누계를 사용한 누적 집계 또는 지정된 기간 내의 레코드를 고려하는 창 집계를 수행할 수 있습니다.

목록 5.5의 코드는 세션 창 내의 트랜잭션 수를 계산합니다. 그림에서. 5.13 이러한 조치는 단계별로 분석됩니다.

windowedBy(SessionWindows.with(twentySeconds).until(fifteenMinutes)) 를 호출하여 비활성 간격이 20초이고 지속 간격이 15분인 세션 창을 만듭니다. 20초의 유휴 간격은 애플리케이션이 현재 세션 종료 또는 시작 후 20초 이내에 도착하는 모든 항목을 현재(활성) 세션에 포함한다는 의미입니다.

다음으로 세션 창에서 수행해야 하는 집계 작업(이 경우 count)을 지정합니다. 수신 항목이 비활성 기간(날짜/시간 스탬프의 양쪽)을 벗어나는 경우 애플리케이션은 새 세션을 생성합니다. 보존 간격은 특정 시간 동안 세션을 유지하는 것을 의미하며 세션의 비활성 기간을 초과하지만 여전히 첨부할 수 있는 지연 데이터를 허용합니다. 또한 병합으로 인한 새 세션의 시작과 끝은 가장 빠른 날짜/시간 스탬프와 가장 늦은 날짜/시간 스탬프에 해당합니다.

세션이 어떻게 작동하는지 알아보기 위해 count 메소드의 몇 가지 항목을 살펴보겠습니다(표 5.1).

레코드가 도착하면 동일한 키, 현재 날짜/타임스탬프보다 작은 종료 시간 - 비활성 간격, 현재 날짜/타임스탬프 + 비활성 간격보다 큰 시작 시간을 가진 기존 세션을 찾습니다. 이를 고려하여 표에 5.1개의 항목이 있습니다. XNUMX은 다음과 같이 단일 세션으로 병합됩니다.

1. 레코드 1이 먼저 도착하므로 시작 시간은 종료 시간과 동일하며 00:00:00입니다.

2. 다음으로 항목 2가 도착하고 23:59:55 이전에 종료되지 않고 00:00:35 이전에 시작되는 세션을 찾습니다. 레코드 1을 찾고 세션 1과 2를 결합합니다. 세션 1의 시작 시간(이전)과 세션 2의 종료 시간(나중)을 사용하여 새 세션이 00:00:00에 시작하고 00에 끝나도록 합니다. 00:15.

3. 레코드 3이 도착하면 00:00:30에서 00:01:10 사이의 세션을 찾았지만 아무것도 찾지 못했습니다. 123:345:654에 시작하고 끝나는 키 00-00-50,FFBE에 대한 두 번째 세션을 추가합니다.

4. 레코드 4가 도착하고 23:59:45에서 00:00:25 사이의 세션을 찾고 있습니다. 이번에는 세션 1과 2가 모두 발견되었으며, 세 세션이 모두 하나로 결합되었으며 시작 시간은 00:00:00, 종료 시간은 00:00:15입니다.

이 섹션에 설명된 내용에서 다음과 같은 중요한 뉘앙스를 기억해 두는 것이 좋습니다.

• 세션은 고정된 크기의 창이 아닙니다. 세션 기간은 주어진 기간 내의 활동에 따라 결정됩니다.
• 데이터의 날짜/타임 스탬프는 이벤트가 기존 세션 내에 속하는지 아니면 유휴 기간에 속하는지 여부를 결정합니다.

다음으로 다음 유형의 창인 "텀블링" 창에 대해 설명하겠습니다.

"텀블링" 창

텀블링 윈도우는 특정 기간 내에 발생하는 이벤트를 캡처합니다. 특정 회사의 모든 주식 거래를 20초마다 캡처해야 하므로 해당 기간 동안의 모든 이벤트를 수집한다고 가정해 보겠습니다. 20초 간격이 끝나면 창이 롤오버되어 새로운 20초 관찰 간격으로 이동합니다. 그림 5.14는 이러한 상황을 보여줍니다.

보시다시피 지난 20초 동안 수신된 모든 이벤트가 창에 포함됩니다. 이 기간이 끝나면 새 창이 생성됩니다.

목록 5.6은 텀블링 윈도우를 사용하여 20초마다 주식 거래를 캡처하는 방법을 보여주는 코드를 보여줍니다(src/main/java/bbejeck/chapter_5/CountingWindowingAndKtableJoinExample.java에 있음).

TimeWindows.of 메서드 호출을 조금만 변경하면 연속 창을 사용할 수 있습니다. 이 예제에서는 Until() 메서드를 호출하지 않으므로 기본 보존 간격인 24시간이 사용됩니다.

마지막으로 창 옵션의 마지막인 "호핑" 창으로 이동할 시간입니다.

슬라이딩("점프") 창

슬라이딩/호핑 창은 텀블링 창과 유사하지만 약간의 차이가 있습니다. 슬라이딩 창은 최근 이벤트를 처리하기 위해 새 창을 만들기 전에 시간 간격이 끝날 때까지 기다리지 않습니다. 창 기간보다 짧은 대기 간격 후에 새 계산을 시작합니다.

텀블링 창과 점프 창의 차이점을 설명하기 위해 증권 거래소 거래 계산의 예로 돌아가 보겠습니다. 우리의 목표는 여전히 트랜잭션 수를 계산하는 것이지만 카운터를 업데이트하기 전에 전체 시간을 기다리고 싶지는 않습니다. 대신 더 짧은 간격으로 카운터를 업데이트합니다. 예를 들어, 우리는 여전히 20초마다 트랜잭션 수를 계산하지만 그림과 같이 5초마다 카운터를 업데이트합니다. 5.15. 이 경우 데이터가 겹치는 결과 창이 XNUMX개 생성됩니다.

목록 5.7은 슬라이딩 창을 정의하기 위한 코드를 보여줍니다(src/main/java/bbejeck/chapter_5/CountingWindowingAndKtableJoinExample.java에 있음).

advanceBy() 메서드에 대한 호출을 추가하여 텀블링 창을 호핑 창으로 변환할 수 있습니다. 표시된 예에서 저장 간격은 15분입니다.

이 섹션에서는 집계 결과를 기간으로 제한하는 방법을 살펴보았습니다. 특히 이 섹션에서 다음 세 가지 사항을 기억하시기 바랍니다.

• 세션 창의 크기는 기간이 아니라 사용자 활동에 따라 제한됩니다.
• "텀블링" 창은 특정 기간 내의 이벤트에 대한 개요를 제공합니다.
• 창 점프 기간은 고정되어 있지만 자주 업데이트되며 모든 창에 중복되는 항목이 포함될 수 있습니다.

다음으로 연결을 위해 KTable을 다시 KStream으로 변환하는 방법을 알아봅니다.

5.3.3. KStream 및 KTable 개체 연결

4장에서 우리는 두 개의 KStream 객체를 연결하는 것에 대해 논의했습니다. 이제 KTable과 KStream을 연결하는 방법을 배워야 합니다. 이는 다음과 같은 간단한 이유로 필요할 수 있습니다. KStream은 레코드 스트림이고 KTable은 레코드 업데이트 스트림이지만 때로는 KTable의 업데이트를 사용하여 레코드 스트림에 추가 컨텍스트를 추가하고 싶을 수도 있습니다.

증권 거래소 거래 건수에 대한 데이터를 취합하여 관련 업종의 증권 거래소 뉴스와 결합해 보겠습니다. 이미 가지고 있는 코드를 고려하여 이를 달성하기 위해 수행해야 할 작업은 다음과 같습니다.

1. 주식 거래 건수에 대한 데이터가 포함된 KTable 개체를 KStream으로 변환한 후 해당 주식 기호에 해당하는 산업 부문을 나타내는 키로 키를 바꿉니다.
2. 증권 거래소 뉴스가 포함된 주제에서 데이터를 읽는 KTable 개체를 만듭니다. 이 새로운 KTable은 산업 분야별로 분류됩니다.
3. 업종별 증권거래소 거래건수 정보와 뉴스 업데이트를 연결합니다.

이제 이 실행 계획을 구현하는 방법을 살펴보겠습니다.

KTable을 KStream으로 변환

KTable을 KStream으로 변환하려면 다음을 수행해야 합니다.

1. KTable.toStream() 메서드를 호출합니다.
2. KStream.map 메소드를 호출하여 키를 업계 이름으로 바꾼 다음 Windowed 인스턴스에서 TransactionSummary 객체를 검색합니다.

이러한 작업을 다음과 같이 연결합니다(코드는 src/main/java/bbejeck/chapter_5/CountingWindowingAndKtableJoinExample.java 파일에서 찾을 수 있음)(목록 5.8).

KStream.map 작업을 수행하고 있기 때문에 반환된 KStream 인스턴스는 연결에서 사용될 때 자동으로 다시 분할됩니다.

변환 프로세스를 완료했습니다. 다음으로 주식 뉴스를 읽기 위한 KTable 개체를 생성해야 합니다.

주식뉴스용 KTable 생성

다행스럽게도 KTable 객체를 생성하는 데는 단 한 줄의 코드만 필요합니다(코드는 src/main/java/bbejeck/chapter_5/CountingWindowingAndKtableJoinExample.java에서 찾을 수 있음)(목록 5.9).

문자열 Serdes가 설정에 사용되므로 Serde 개체를 지정할 필요가 없다는 점은 주목할 가치가 있습니다. 또한 EARLIEST 열거형을 사용하면 테이블의 맨 처음 부분이 레코드로 채워집니다.

이제 마지막 단계인 연결로 넘어갈 수 있습니다.

뉴스 업데이트를 거래 횟수 데이터와 연결

연결을 만드는 것은 어렵지 않습니다. 관련 산업에 대한 주식 뉴스가 없는 경우 Left 조인을 사용합니다(필요한 코드는 src/main/java/bbejeck/chapter_5/CountingWindowingAndKtableJoinExample.java 파일에서 찾을 수 있음)(목록 5.10).

이 leftJoin 연산자는 매우 간단합니다. 4장의 조인과 달리 KStream-KTable 조인을 수행할 때 각 키에 대해 KTable에 하나의 항목만 있기 때문에 JoinWindow 메서드는 사용되지 않습니다. 이러한 연결은 시간에 제한이 없습니다. 레코드는 KTable에 있거나 없습니다. 주요 결론: KTable 개체를 사용하면 자주 업데이트되지 않는 참조 데이터로 KStream을 강화할 수 있습니다.

이제 KStream에서 이벤트를 풍성하게 하는 보다 효율적인 방법을 살펴보겠습니다.

5.3.4. GlobalKTable 객체

보시다시피 이벤트 스트림을 강화하거나 이벤트 스트림에 컨텍스트를 추가해야 합니다. 4장에서는 두 KStream 객체 사이의 연결을 보았고 이전 섹션에서는 KStream과 KTable 사이의 연결을 보았습니다. 이러한 모든 경우에 키를 새로운 유형이나 값에 매핑할 때 데이터 스트림을 다시 분할해야 합니다. 재분할이 명시적으로 수행되는 경우도 있고 Kafka Streams가 자동으로 수행하는 경우도 있습니다. 키가 변경되었고 레코드가 새 섹션에 있어야 하기 때문에 다시 파티셔닝이 필요합니다. 그렇지 않으면 연결이 불가능합니다(이 내용은 4장 하위 섹션 4.2.4의 "데이터 다시 파티셔닝" 섹션에서 논의되었습니다).

재파티셔닝에는 비용이 듭니다

다시 분할하려면 비용이 필요합니다. 중간 주제를 생성하고 다른 주제에 중복 데이터를 저장하는 데 드는 추가 리소스 비용입니다. 이는 또한 이 주제에 대한 쓰기 및 읽기로 인해 대기 시간이 증가했음을 의미합니다. 또한 둘 이상의 측면이나 차원에 걸쳐 조인해야 하는 경우 조인을 연결하고 레코드를 새 키로 매핑한 다음 다시 파티셔닝 프로세스를 다시 실행해야 합니다.

더 작은 데이터 세트에 연결

어떤 경우에는 연결될 참조 데이터의 양이 상대적으로 작기 때문에 전체 복사본이 각 노드에 로컬로 쉽게 들어갈 수 있습니다. 이와 같은 상황을 위해 Kafka Streams는 GlobalKTable 클래스를 제공합니다.

GlobalKTable 인스턴스는 애플리케이션이 모든 데이터를 각 노드에 복제하기 때문에 고유합니다. 그리고 모든 데이터가 각 노드에 존재하므로 모든 파티션에서 사용할 수 있도록 참조 데이터 키로 이벤트 스트림을 분할할 필요가 없습니다. GlobalKTable 개체를 사용하여 키 없는 조인을 만들 수도 있습니다. 이 기능을 보여주기 위해 이전 예제 중 하나로 돌아가 보겠습니다.

KStream 개체를 GlobalKTable 개체에 연결

하위 섹션 5.3.2에서는 구매자별 교환 거래 창 집계를 수행했습니다. 이 집계의 결과는 다음과 같습니다.

{customerId='074-09-3705', stockTicker='GUTM'}, 17
{customerId='037-34-5184', stockTicker='CORK'}, 16

이러한 결과는 목적에 부합하지만 고객 이름과 전체 회사 이름도 표시되었으면 더 유용했을 것입니다. 고객 이름과 회사 이름을 추가하려면 일반 조인을 수행할 수 있지만 두 가지 키 매핑과 다시 파티셔닝을 수행해야 합니다. GlobalKTable을 사용하면 이러한 작업 비용을 피할 수 있습니다.

이를 위해 목록 5.11의 countStream 개체(해당 코드는 src/main/java/bbejeck/chapter_5/GlobalKTableExample.java에서 찾을 수 있음)를 사용하고 이를 두 개의 GlobalKTable 개체에 연결합니다.

이에 대해서는 이미 이전에 논의했으므로 반복하지 않겠습니다. 하지만 toStream().map 함수의 코드는 가독성을 위해 인라인 람다 표현식 대신 함수 개체로 추상화되어 있습니다.

다음 단계는 GlobalKTable의 두 인스턴스를 선언하는 것입니다(표시된 코드는 src/main/java/bbejeck/chapter_5/GlobalKTableExample.java 파일에서 찾을 수 있음)(목록 5.12).

주제 이름은 열거 유형을 사용하여 설명됩니다.

이제 모든 구성 요소가 준비되었으므로 남은 것은 연결을 위한 코드(src/main/java/bbejeck/chapter_5/GlobalKTableExample.java 파일에서 찾을 수 있음)를 작성하는 것뿐입니다(목록 5.13).

이 코드에는 두 개의 조인이 있지만 결과가 별도로 사용되지 않기 때문에 연결되어 있습니다. 전체 작업이 끝나면 결과가 표시됩니다.

위의 조인 작업을 실행하면 다음과 같은 결과를 얻게 됩니다.

{customer='Barney, Smith' company="Exxon", transactions= 17}

본질은 변하지 않았지만 이러한 결과는 더욱 명확해 보입니다.

4장까지 카운트다운하면 이미 여러 유형의 연결이 작동하는 것을 볼 수 있습니다. 그것들은 표에 나열되어 있습니다. 5.2. 이 표는 Kafka Streams 버전 1.0.0의 연결 기능을 반영합니다. 향후 릴리스에서는 내용이 변경될 수 있습니다.

마무리하기 위해 기본 사항을 요약해 보겠습니다. 로컬 상태를 사용하여 이벤트 스트림(KStream)과 업데이트 스트림(KTable)을 연결할 수 있습니다. 또는 참조 데이터의 크기가 너무 크지 않으면 GlobalKTable 개체를 사용할 수 있습니다. GlobalKTables는 모든 파티션을 각 Kafka Streams 애플리케이션 노드에 복제하여 키가 해당하는 파티션에 관계없이 모든 데이터를 사용할 수 있도록 보장합니다.

다음으로 Kafka 주제의 데이터를 소비하지 않고도 상태 변경을 관찰할 수 있는 Kafka Streams 기능을 살펴보겠습니다.

5.3.5. 쿼리 가능한 상태

우리는 이미 상태와 관련된 여러 작업을 수행했으며 항상 결과를 콘솔에 출력하거나(개발 목적으로) 주제에 기록합니다(프로덕션 목적으로). 주제에 결과를 쓸 때 Kafka 소비자를 사용하여 결과를 확인해야 합니다.

이러한 주제의 데이터를 읽는 것은 일종의 구체화된 뷰로 간주될 수 있습니다. 우리의 목적을 위해 Wikipedia의 구체화된 뷰 정의를 사용할 수 있습니다. “...쿼리 결과를 포함하는 물리적 데이터베이스 객체. 예를 들어, 원격 데이터의 로컬 복사본일 수도 있고, 테이블이나 조인 결과의 행 및/또는 열의 하위 집합일 수도 있고, 집계를 통해 얻은 요약 테이블일 수도 있습니다." (https://en.wikipedia.org/wiki) /Materialized_view).

Kafka Streams를 사용하면 상태 저장소에서 대화형 쿼리를 실행할 수 있으므로 이러한 구체화된 뷰를 직접 읽을 수 있습니다. 상태 저장소에 대한 쿼리는 읽기 전용 작업이라는 점에 유의하는 것이 중요합니다. 이렇게 하면 애플리케이션이 데이터를 처리하는 동안 실수로 상태가 불일치하게 되는 것에 대해 걱정할 필요가 없습니다.

상태 저장소를 직접 쿼리하는 기능이 중요합니다. 이는 먼저 Kafka 소비자로부터 데이터를 가져올 필요 없이 대시보드 애플리케이션을 생성할 수 있음을 의미합니다. 또한 데이터를 다시 쓸 필요가 없기 때문에 애플리케이션의 효율성도 높아집니다.

• 데이터의 지역성 덕분에 빠르게 액세스할 수 있습니다.
• 외부 저장소에 기록되지 않으므로 데이터 중복이 제거됩니다.

기억해두셨으면 하는 가장 중요한 점은 애플리케이션 내에서 상태를 직접 쿼리할 수 있다는 것입니다. 이것이 제공하는 기회는 아무리 강조해도 지나치지 않습니다. Kafka의 데이터를 사용하고 애플리케이션용 데이터베이스에 레코드를 저장하는 대신 동일한 결과로 상태 저장소를 쿼리할 수 있습니다. 상태 저장소에 대한 직접 쿼리는 더 적은 코드(소비자 없음)와 더 적은 소프트웨어(결과를 저장하기 위한 데이터베이스 테이블이 필요 없음)를 의미합니다.

이 장에서 꽤 많은 내용을 다루었으므로 지금은 상태 저장소에 대한 대화형 쿼리에 대한 논의를 그대로 두겠습니다. 하지만 걱정하지 마세요. 9장에서는 대화형 쿼리가 포함된 간단한 대시보드 애플리케이션을 만들어 보겠습니다. 이 장과 이전 장의 예제 중 일부를 사용하여 대화형 쿼리와 이를 Kafka Streams 애플리케이션에 추가하는 방법을 보여줍니다.

개요

• KStream 개체는 데이터베이스에 삽입하는 것과 비슷한 이벤트 스트림을 나타냅니다. KTable 개체는 데이터베이스 업데이트와 유사한 업데이트 스트림을 나타냅니다. KTable 개체의 크기는 커지지 않으며 이전 레코드는 새 레코드로 대체됩니다.
• 집계 작업에는 KTable 개체가 필요합니다.
• 기간 작업을 사용하면 집계된 데이터를 시간 버킷으로 분할할 수 있습니다.
• GlobalKTable 개체 덕분에 파티셔닝에 관계없이 애플리케이션 어디에서나 참조 데이터에 액세스할 수 있습니다.
• KStream, KTable 및 GlobalKTable 개체 간의 연결이 가능합니다.

지금까지 우리는 높은 수준의 KStream DSL을 사용하여 Kafka Streams 애플리케이션을 구축하는 데 중점을 두었습니다. 높은 수준의 접근 방식을 사용하면 깔끔하고 간결한 프로그램을 만들 수 있지만 이를 사용하는 것은 절충안이 됩니다. DSL KStream으로 작업한다는 것은 제어 수준을 줄여 코드의 간결성을 높이는 것을 의미합니다. 다음 장에서는 하위 수준 핸들러 노드 API를 살펴보고 다른 절충안을 시도해 보겠습니다. 프로그램은 이전보다 길어지지만 필요한 거의 모든 핸들러 노드를 만들 수 있습니다.

→ 책에 대한 자세한 내용은 다음에서 확인할 수 있습니다. 발행인의 웹사이트

→ 쿠폰을 이용하여 Habrozhiteli 25% 할인을 받으려면 - 카프카 스트림

→ 종이책 결제 시, 전자책이 이메일로 발송됩니다.

출처 : habr.com