[java] Kafka Streams와 Amazon Redshift와의 통합 방법

16 Nov 2023

java

이번 글에서는 Kafka Streams와 Amazon Redshift를 연동하는 방법에 대해 알아보겠습니다.

1. Kafka Streams 소개

Kafka Streams는 Apache Kafka를 기반으로 하는 스트리밍 데이터 처리 라이브러리입니다. 이를 사용하면 실시간으로 데이터를 처리하고 분석하는 애플리케이션을 쉽게 개발할 수 있습니다.

2. Amazon Redshift 소개

Amazon Redshift는 클라우드 기반의 데이터 웨어하우스 솔루션으로, 대용량의 데이터를 고속으로 저장하고 분석할 수 있습니다. 이를 활용하면 대규모 데이터를 실시간으로 쿼리하여 의사결정에 활용할 수 있습니다.

3. Kafka Streams와 Amazon Redshift 연동 방법

Kafka Streams와 Amazon Redshift를 연동하기 위해서는 다음과 같은 단계를 따라야 합니다.

3.1. Kafka Streams에서 데이터 처리

Kafka Streams를 사용하여 데이터를 처리하는 코드를 작성합니다. 이 코드에서는 Kafka 토픽에서 데이터를 소비하고, 원하는 처리를 수행한 뒤 적당한 형식으로 Redshift에 저장합니다. 예를 들어, 데이터를 정제하거나 집계하는 등의 작업을 수행할 수 있습니다.

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> stream = builder.stream("my-topic");

KTable<String, Long> counts = stream
    .flatMapValues(value -> Arrays.asList(value.toLowerCase().split("\\W+")))
    .groupBy((key, word) -> word)
    .count();

counts.toStream().to("my-output-topic", Produced.with(Serdes.String(), Serdes.Long()));

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

3.2. Redshift에서 데이터를 저장

Amazon Redshift에서 데이터를 저장하기 위해 데이터를 적절한 형식으로 변환하여 저장합니다. 이를 위해서는 JDBC 드라이버를 사용하여 Redshift와 연결한 뒤 데이터를 삽입하는 작업을 수행합니다.

try (Connection conn = DriverManager.getConnection(dbUrl, user, password);
     Statement stmt = conn.createStatement()) {
  
    String sql = "INSERT INTO my_table (word, count) VALUES (?, ?)";
    PreparedStatement pstmt = conn.prepareStatement(sql);

    for (KeyValue<String, Long> record : records) {
        pstmt.setString(1, record.key);
        pstmt.setLong(2, record.value);
        pstmt.addBatch();
    }

    pstmt.executeBatch();
}

4. 결론

Kafka Streams와 Amazon Redshift를 연동하여 스트리밍 데이터를 실시간으로 처리하고 분석하는 것은 강력한 데이터 파이프라인을 구축하기 위한 좋은 방법입니다. 이를 통해 데이터 기반의 의사결정을 신속하고 효과적으로 수행할 수 있습니다.

참고자료: