[java] 자바로 카프카와 함께하는 실시간 데이터 분석

16 Nov 2023

java

이번 블로그 포스트에서는 자바를 사용하여 카프카와 함께 실시간 데이터 분석을 어떻게 수행하는지 알아보겠습니다. 카프카는 분산형 스트리밍 플랫폼으로, 대용량의 실시간 데이터를 처리할 수 있습니다. 또한, 자바는 많은 기업에서 사용되는 프로그래밍 언어로, 데이터 분석을 위한 다양한 라이브러리와 도구를 제공합니다.

카프카와 자바 연동

카프카와 자바를 연동하기 위해서는 먼저 카프카 클라이언트를 자바 프로젝트에 추가해야 합니다. Maven을 사용하는 경우, pom.xml 파일에 다음 의존성을 추가합니다.

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>2.8.1</version>
</dependency>

위의 의존성을 추가하면, 자바 프로젝트에서 카프카 클라이언트를 사용할 수 있게 됩니다.

카프카에서 데이터 수신하기

카프카로부터 데이터를 수신하기 위해서는 KafkaConsumer 클래스를 사용합니다. 다음은 간단한 예제입니다.

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import java.util.Arrays;
import java.util.Properties;

public class KafkaConsumerExample {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("group.id", "my-group");
        props.put("key.deserializer",
                "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer",
                "org.apache.kafka.common.serialization.StringDeserializer");

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Arrays.asList("my-topic"));

        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(100);
            for (ConsumerRecord<String, String> record : records) {
                System.out.printf("Received message: key = %s, value = %s%n",
                        record.key(), record.value());
            }
        }
    }
}

위의 코드에서는 bootstrap.servers를 카프카 브로커의 주소로 설정하고, group.id를 컨슈머 그룹의 식별자로 설정합니다. 그리고 subscribe 메서드를 사용하여 구독할 토픽을 지정합니다.

데이터 분석하기

카프카로부터 데이터를 수신한 후, 이를 분석하는 작업을 수행할 수 있습니다. 자바에서는 다양한 라이브러리와 도구를 사용하여 데이터 분석을 할 수 있습니다. 예를 들어, Apache Spark와 같은 분산 데이터 처리 프레임워크를 사용하여 카프카로부터 데이터를 읽고 분석할 수 있습니다.

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.streaming.StreamingQuery;
import org.apache.spark.sql.streaming.StreamingQueryException;

public class KafkaDataAnalysis {
    public static void main(String[] args) throws StreamingQueryException {
        SparkSession spark = SparkSession.builder()
                .appName("KafkaDataAnalysis")
                .master("local[*]")
                .getOrCreate();

        Dataset<Row> df = spark
                .readStream()
                .format("kafka")
                .option("kafka.bootstrap.servers", "localhost:9092")
                .option("subscribe", "my-topic")
                .load();

        df.printSchema();

        StreamingQuery query = df
                .writeStream()
                .format("console")
                .start();

        query.awaitTermination();
    }
}

위의 코드에서는 SparkSession을 사용하여 Spark를 초기화하고, readStream 메서드를 사용하여 카프카로부터 데이터를 읽습니다. 그리고 writeStream 메서드를 사용하여 데이터를 출력할 위치를 지정합니다. 이 예제에서는 데이터를 콘솔에 출력하도록 설정하였습니다.

결론

이번 블로그 포스트에서는 자바를 사용하여 카프카와 함께 실시간 데이터 분석을 어떻게 수행하는지 알아보았습니다. 카프카의 강력한 분산 메시징 기능과 자바의 다양한 라이브러리와 도구를 활용하여 실시간 데이터 분석을 수행할 수 있습니다. 추가적으로, 다른 라이브러리나 도구를 사용하여 더 복잡한 데이터 분석 작업을 수행할 수도 있습니다.

카프카와 자바 연동

카프카에서 데이터 수신하기

데이터 분석하기

결론

참고 자료