카프카 핵심 가이드 14장 - 스트림 처리

// 카프카 스트림즈를 설정 // APPLICATION_ID_CONFIG: 스트림 애플리케이션의 고유 식별자 // BOOTSTRAP_SERVERS_CONFIG: 카프카 클러스터 연결 주소 // DEFAULT_KEY_SERDE_CLASS_CONFIG: 키의 Serde(Serializer/Deserializer) 클래스 // DEFAULT_VALUE_SERDE_CLASS_CONFIG: 값의 Serde 클래스 public class WordCountExample { public static void main(String[] args) throws Exception { Properties props = new Properties(); // 애플리케이션 ID 설정 props.put(StreamsConfig.APPLICATION_ID_CONFIG, "wordcount"); // 카프카 브로커 주소 설정 props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092"); // 키의 직렬화/역직렬화 클래스 설정 props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName()); // 값의 직렬화/역직렬화 클래스 설정 props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName()); } } // 스트림즈 토폴로지를 생성 // 1. 스트림 빌더 및 소스 스트림 생성 StreamsBuilder builder = new StreamsBuilder(); KStream<String, String> source = builder.stream("wordcount-input"); // 2. 단어 분리 패턴 정의 final Pattern pattern = Pattern.compile("\\W+"); // 3. 스트림 처리 파이프라인 구성 KStream<String, String> counts = source // 문자열을 소문자로 변환하고 단어로 분리 .flatMapValues(value -> Arrays.asList(pattern.split(value.toLowerCase()))) // 각 단어를 키로 설정 .map((key, value) -> new KeyValue<String, String>(value, value)) // "the" 단어 필터링 .filter((key, value) -> (!value.equals("the"))) // 키별로 그룹화 .groupByKey() // 개수 세기 .count() // Long을 String으로 변환 .mapValues(value -> Long.toString(value)) // KTable을 KStream으로 변환 .toStream(); // 4. 결과를 출력 토픽에 저장 counts.to("wordcount-output"); // 애플리케이션이 수행할 변환의 흐름을 정의했으므로, 실행 // 5. 스트림 처리 실행 KafkaStreams streams = new KafkaStreams(builder.build(), props); streams.start(); // 예제를 위한 임시 실행(5초) Thread.sleep(5000L); streams.close();