Kafka Streams와 스프링 부트 통합: 실시간 데이터 처리 애플리케이션 구축하기

 

Apache Kafka Streams 는 실시간 데이터 스트리밍을 처리하는 강력한 API로, Kafka의 데이터를 실시간으로 처리하고 변환할 수 있습니다. 이번 포스팅에서는 Kafka Streams API 와 스프링 부트 를 통합하여 실시간 데이터 처리 애플리케이션 을 구축하는 방법을 설명합니다. 또한 상태 저장 변환 과 인터랙티브 쿼리 를 통해 애플리케이션에서 상태 저장소 를 어떻게 활용하는지도 살펴봅니다.

 

---

 

 

1. Kafka Streams API 소개

Kafka Streams 는 Apache Kafka 위에서 동작하는 스트림 처리 라이브러리 로, 데이터를 실시간으로 변환, 필터링, 집계 할 수 있습니다. 특히 분산 환경 에서 동작할 수 있어 대규모 데이터 스트리밍 처리에 적합합니다.

Kafka Streams의 주요 기능

1. 실시간 데이터 처리 : 메시지를 실시간으로 변환하고 처리할 수 있습니다.
2. 분산 시스템 : Kafka 클러스터 내에서 확장 가능하게 동작합니다.
3. 상태 저장 : 상태 저장소 를 통해 데이터의 집계 나 변환 결과 를 메모리나 디스크에 저장할 수 있습니다.

예시: Kafka Streams로 단순 메시지 필터링

StreamsBuilder builder = new StreamsBuilder();
KStream<String, String> stream = builder.stream("input_topic");

KStream<String, String> filteredStream = stream.filter((key, value) -> value.contains("Kafka"));

filteredStream.to("output_topic");

 

---

 

2. Kafka Streams 애플리케이션 설정

Kafka Streams를 스프링 부트와 통합하려면 Spring Kafka 를 사용하여 설정할 수 있습니다. 스프링 부트에서는 Kafka Streams 를 손쉽게 설정할 수 있도록 다양한 기능을 제공합니다.

스프링 부트 설정 예시 ( application.yml )

spring:
  kafka:
    streams:
      application-id: my-kafka-streams-app
      bootstrap-servers: localhost:9092
      default-key-serde: org.apache.kafka.common.serialization.Serdes$StringSerde
      default-value-serde: org.apache.kafka.common.serialization.Serdes$StringSerde

Kafka Streams 애플리케이션 예시

@EnableKafkaStreams
@SpringBootApplication
public class KafkaStreamsApp {

    public static void main(String[] args) {
        SpringApplication.run(KafkaStreamsApp.class, args);
    }

    @Bean
    public KStream<String, String> kStream(StreamsBuilder streamsBuilder) {
        KStream<String, String> stream = streamsBuilder.stream("input_topic");

        KStream<String, String> uppercasedStream = stream.mapValues(value -> value.toUpperCase());
        uppercasedStream.to("output_topic");

        return uppercasedStream;
    }
}

설명:

• @EnableKafkaStreams : Kafka Streams 애플리케이션을 활성화합니다.
• StreamsBuilder : 스트림을 생성하고 처리하는 API.
• mapValues() : 각 메시지의 값을 대문자로 변환하는 간단한 예시입니다.

 

---

 

3. 상태 저장 변환 수행하기

Kafka Streams는 상태 저장 변환 을 통해 데이터를 집계 하거나 변환 한 결과를 로컬 상태 저장소 에 저장할 수 있습니다. 이러한 상태 저장은 세션 , 집계 와 같은 고급 스트림 처리를 가능하게 합니다.

상태 저장 예시: 메시지 집계

KGroupedStream<String, String> groupedStream = stream.groupByKey();
KTable<String, Long> aggregatedTable = groupedStream.count(Materialized.as("counts-store"));

aggregatedTable.toStream().to("aggregated_output_topic");

설명:

• groupByKey() : 키를 기준으로 데이터를 그룹화합니다.
• count() : 각 키의 발생 횟수를 집계합니다.
• Materialized.as() : 집계 결과를 상태 저장소 에 저장합니다.

 

---

 

4. 인터랙티브 쿼리와 상태 저장소 활용

Kafka Streams 의 강력한 기능 중 하나는 상태 저장소 에 저장된 데이터를 인터랙티브 쿼리 를 통해 실시간으로 조회할 수 있다는 점입니다. 예를 들어, 특정 키의 집계 결과를 API로 제공할 수 있습니다.

상태 저장소에서 쿼리 수행 예시

@RestController
public class QueryController {

    @Autowired
    private QueryableStoreRegistry queryableStoreRegistry;

    @GetMapping("/count/{key}")
    public Long getCount(@PathVariable String key) {
        ReadOnlyKeyValueStore<String, Long> keyValueStore =
                queryableStoreRegistry.getQueryableStoreType("counts-store", QueryableStoreTypes.keyValueStore());

        return keyValueStore.get(key);
    }
}

설명:

• QueryableStoreRegistry : Kafka Streams에서 상태 저장소에 쿼리할 수 있는 API입니다.
• ReadOnlyKeyValueStore : 상태 저장소에서 특정 키의 데이터를 읽어오는 API입니다.

 

---

 

Kafka Streams와 스프링 부트 통합 트렌드 및 통계

최근 데이터에 따르면, Kafka Streams 를 사용한 실시간 데이터 처리 애플리케이션은 기업에서 데이터 분석 , 모니터링 , 실시간 대시보드 구축 등에 널리 사용되고 있습니다. 실시간 스트리밍 데이터 처리 를 도입한 기업의 비즈니스 의사결정 속도 는 평균적으로 30% 이상 향상 되었다고 보고됩니다.

 

---

 

자주 묻는 질문 (FAQ)

Q1. Kafka Streams와 스프링 부트를 통합하려면 어떤 설정이 필요한가요?
A1. Spring Kafka 설정에서 Kafka Streams 관련 옵션을 설정하고, @EnableKafkaStreams 를 사용하여 Kafka Streams 애플리케이션을 활성화할 수 있습니다.

Q2. Kafka Streams에서 상태 저장 변환은 어떻게 수행되나요?
A2. Materialized.as() 를 사용하여 변환 결과를 상태 저장소 에 저장할 수 있습니다. 상태 저장 변환은 집계, 윈도우 처리 등의 복잡한 스트림 처리를 가능하게 합니다.

Q3. 인터랙티브 쿼리는 무엇인가요?
A3. 인터랙티브 쿼리 는 Kafka Streams의 상태 저장소 에 저장된 데이터를 API를 통해 실시간으로 조회할 수 있는 기능입니다. 이를 통해 애플리케이션 외부에서 상태 저장된 데이터를 활용할 수 있습니다.

Q4. Kafka Streams 애플리케이션에서 상태 저장소를 어떻게 활용할 수 있나요?
A4. 상태 저장소는 세션 유지 , 데이터 집계 , 변환 결과 저장 등 다양한 용도로 사용됩니다. 쿼리 를 통해 실시간으로 저장된 데이터를 조회할 수도 있습니다.

Q5. Kafka Streams와 스프링 부트를 사용하는 주된 이유는 무엇인가요?
A5. Kafka Streams와 스프링 부트의 통합은 실시간 데이터 처리 를 간편하게 구현할 수 있는 강력한 솔루션을 제공합니다. 스프링 부트의 유연한 설정과 Kafka Streams의 강력한 스트림 처리 기능이 결합되어 대규모 실시간 데이터 애플리케이션을 쉽게 구축할 수 있습니다.