[java] 자바로 카프카를 활용한 로그 수집 및 분석

16 Nov 2023

java

소개

카프카(Kafka)는 대용량의 실시간 데이터를 안정적으로 처리하기 위한 분산 스트리밍 플랫폼입니다. 여러 시스템 간의 데이터 플로우를 처리하고, 데이터의 신뢰성, 확장성 및 내구성을 보장하는데 사용됩니다. 이러한 특징으로 인해 카프카는 로그 수집, 실시간 분석, 활용 등 다양한 용도로 사용되고 있습니다.

카프카 로그 수집 및 분석 프로세스

1. 카프카 설치 및 설정

먼저, 카프카를 설치하고 클러스터를 설정해야 합니다. 아파치 카프카 공식 홈페이지에서 다운로드하고, 필요한 설정을 진행합니다.

2. 로그 프로듀서 작성

로그를 생성해 카프카에 전달하는 로그 프로듀서를 작성해야 합니다. 자바로 개발하려면 KafkaProducer 클래스를 사용하여 카프카로 메시지를 전송할 수 있습니다. 메시지를 생성하고 ProducerRecord로 래핑하여 카프카의 토픽에 메시지를 보내면 됩니다.

예시 코드:

import org.apache.kafka.clients.producer.*;

public class LogProducer {
    public static void main(String[] args) {
        // 카프카 클러스터 설정
        String bootstrapServers = "localhost:9092";

        // 프로듀서 설정
        Properties properties = new Properties();
        properties.put("bootstrap.servers", bootstrapServers);
        properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        // 프로듀서 생성
        KafkaProducer<String, String> producer = new KafkaProducer<>(properties);

        // 메시지 전송
        String topic = "logs";
        String message = "This is a log message";
        ProducerRecord<String, String> record = new ProducerRecord<>(topic, message);
        producer.send(record);

        // 프로듀서 종료
        producer.close();
    }
}

3. 로그 컨슈머 작성

카프카에서 메시지를 소비하여 로그를 분석하고 저장하는 로그 컨슈머를 작성해야 합니다. 로그 컨슈머는 KafkaConsumer 클래스를 사용하여 카프카로부터 메시지를 받아옵니다. 받아온 메시지를 분석하여 필요한 작업을 수행한 뒤 로그를 저장합니다.

예시 코드:

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

public class LogConsumer {
    public static void main(String[] args) {
        // 카프카 클러스터 설정
        String bootstrapServers = "localhost:9092";

        // 컨슈머 설정
        Properties properties = new Properties();
        properties.put("bootstrap.servers", bootstrapServers);
        properties.put("group.id", "log-consumer-group");
        properties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        properties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

        // 컨슈머 생성
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties);

        // 토픽 구독
        String topic = "logs";
        consumer.subscribe(Collections.singletonList(topic));

        // 메시지 소비
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
            for (ConsumerRecord<String, String> record : records) {
                // 로그 분석 및 저장 작업 수행
                System.out.println(record.value());
            }
            consumer.commitSync();
        }
    }
}

결론

자바를 사용하여 카프카를 활용한 로그 수집 및 분석 프로세스를 구현할 수 있습니다. 카프카의 확장성과 유연성을 활용하여 대용량의 로그 데이터를 실시간으로 처리하고 분석할 수 있습니다. 자세한 내용은 카프카 공식 문서를 참조하시기 바랍니다.