[java] Kafka Streams와 ELK 스택과의 통합 방법

16 Nov 2023

java

이 문서에서는 Kafka Streams와 ELK (Elasticsearch, Logstash, Kibana) 스택을 사용하여 데이터를 처리하고 시각화하는 방법을 설명합니다.

개요

Kafka Streams는 Apache Kafka에서 제공하는 라이브러리로, 고성능의 실시간 스트리밍 데이터 처리를 위해 설계되었습니다. ELK 스택은 Elasticsearch, Logstash, Kibana를 함께 사용하여 로그 데이터를 수집, 처리하고 시각화하기 위한 도구들의 집합입니다.

Kafka Streams와 ELK 스택을 함께 사용하면 데이터를 실시간으로 처리하고, 데이터의 가시성과 분석을 위해 Kibana를 사용할 수 있습니다.

Kafka Streams 개요

Kafka Streams는 Kafka의 Consumer와 Producer API를 기반으로하는 라이브러리입니다. 이를 사용하여 Kafka에서 데이터를 읽어들이고, 데이터를 처리하고, 다시 Kafka로 데이터를 쓸 수 있습니다. Kafka Streams는 내장된 스트림 처리 엔진을 사용하며, 이를 통해 데이터를 높은 처리량과 낮은 지연 시간으로 처리할 수 있습니다.

ELK 스택 개요

ELK 스택은 Elasticsearch, Logstash, Kibana를 함께 사용하여 로그 데이터를 수집, 처리 및 시각화합니다.

Elasticsearch: 대용량의 실시간 검색 및 분석을 위한 오픈 소스 분산 검색 엔진입니다.
Logstash: 다양한 소스로부터 데이터를 수집하고 필터링하고 변환한 다음, Elasticsearch로 전송하는 오픈 소스 데이터 수집 엔진입니다.
Kibana: Elasticsearch에서 수집된 데이터를 시각화하고 대시보드를 만들 수 있는 오픈 소스 데이터 시각화 도구입니다.

Kafka Streams와 ELK 스택 통합 방법

Kafka Streams와 ELK 스택을 통합하기 위해 다음 단계를 따를 수 있습니다.

Kafka Streams 애플리케이션에서 데이터를 처리하여 원하는 형식으로 변환합니다.
이 변환된 데이터를 Logstash를 통해 Elasticsearch로 전송합니다.
Elasticsearch에 저장된 데이터를 Kibana를 사용하여 시각화하고 대시보드를 만듭니다.

아래는 Kafka Streams 애플리케이션에서 변환된 데이터를 Logstash로 전송하는 예제 코드입니다.

Properties props = new Properties();
props.put(StreamsConfig.APPLICATION_ID_CONFIG, "my-streams-app");
props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());

StreamsBuilder builder = new StreamsBuilder();
KStream<String, String> inputStream = builder.stream("input-topic");

// 데이터 처리 및 변환 작업
KStream<String, String> transformedStream = inputStream.mapValues(value -> {
    // 데이터 처리 로직 작성
    String transformedValue = // 변환된 데이터 생성
    return transformedValue;
});

// 변환된 데이터를 Logstash로 전송
transformedStream.to("output-topic");

KafkaStreams streams = new KafkaStreams(builder.build(), props);
streams.start();

위의 예제 코드에서는 Kafka Streams 애플리케이션에서 입력 토픽에서 데이터를 읽어들이고, 데이터 처리 및 변환 작업을 수행한 후 변환된 데이터를 출력 토픽으로 전송합니다.

이제 Logstash를 사용하여 토픽에서 데이터를 읽어들여 Elasticsearch로 전송할 수 있습니다.

input {
  kafka {
    bootstrap_servers => "localhost:9092"
    topics => ["output-topic"]
    group_id => "my-logstash-consumer-group"
  }
}

output {
  elasticsearch {
    hosts => ["localhost:9200"]
    index => "my-index"
  }
}

위의 Logstash 설정 파일 예제에서는 Kafka 토픽인 “output-topic”에서 데이터를 읽어들여 Elasticsearch의 “my-index” 인덱스에 저장합니다.

마지막으로, Kibana를 사용하여 Elasticsearch에 저장된 데이터를 시각화하고 대시보드를 만들 수 있습니다.

결론

Kafka Streams와 ELK 스택을 통합하면 데이터를 처리하고 시각화할 수 있는 강력한 데이터 기술 스택을 구축할 수 있습니다. 이를 통해 실시간으로 데이터를 처리 및 분석하여 의사 결정에 활용할 수 있습니다.

목차

개요

Kafka Streams 개요

ELK 스택 개요

Kafka Streams와 ELK 스택 통합 방법

결론