Kafka와 AI로 콘텐츠 자동 수집 및 분류 시스템 만들기

 

DevNote

1. 왜 이벤트 기반 데이터 파이프라인인가?

devnote 프로젝트의 핵심은 외부 콘텐츠를 가져와 가공하는 것이다. 만약 데이터 수집, AI 분류, DB 저장을 하나의 동기적 흐름으로 짰다면 AI 분류 API가 느려지거나 실패했을 때 전체 데이터 수집 프로세스가 멈춰버리는 문제가 발생했을 것이다. 이러한 문제를 해결하고 각 기능의 독립성을 보장하기 위해, Kafka를 이용한 이벤트 기반 비동기 데이터 파이프라인을 구축했다.

 

2. 전체 데이터 파이프라인 아키텍처

데이터의 흐름은 명확한 역할 분담을 따르도록 설계했다.

 

1. 데이터 생산 (Produce): news-youtube-service가 외부 소스에서 데이터를 수집하여 Kafka 토픽으로 발행한다.

2. 메시지 브로커 (Broker): Kafka가 중간에서 메시지를 보관하고 전달하는 역할을 한다.

3. 데이터 소비 (Consume): processor-service가 Kafka 토픽의 메시지를 구독하여 가져온 뒤 데이터를 가공하고 최종적으로 저장한다.

 

외부 소스 → News-Youtube-Service → Kafka (raw.content 토픽) → Processor-Service → MariaDB / Elasticsearch

 

3. 데이터 수집 및 발행 (news-youtube-service)

이 서비스의 책임은 "외부 데이터를 수집하여 Kafka에 전달하는 것" 이었다. application.yml에 정의된 뉴스 RSS 피드와 유튜브 채널 목록을 기반으로, @Scheduled 스케줄러가 주기적으로 데이터를 수집했다.

수집된 모든 데이터는 표준화된 ContentMessageDto로 변환된 후, raw.content라는 이름의 Kafka 토픽으로 발행되었다. 이 서비스의 역할은 데이터를 발행하는 순간 끝난다.

 

4. 데이터 소비, 가공, 저장 (processor-service)

processor-service는 데이터 파이프라인의 핵심 처리 허브 역할을 수행했다.

 

4-1. Kafka 메시지 소비

ContentService 내의 @KafkaListener가 raw.content 토픽을 실시간으로 구독하여 새로운 콘텐츠 데이터를 받아왔다.

4-2. AI 기반 자동 분류

받아온 데이터의 category 필드가 'TBC(To Be Classified)' 상태일 경우, CategoryClassificationService를 통해 Google Gemini AI API를 호출했다. AI는 콘텐츠의 제목을 분석하여 '백엔드', '보안' 등 가장 적합한 카테고리를 반환하도록 했다.

4-3. Polyglot Persistence (이중 저장)

최종적으로 가공된 데이터는 두 곳에 저장했다. 안정적인 트랜잭션 관리가 필요한 원본 데이터는 MariaDB에, 그리고 검색 기능을 위해서는 Elasticsearch에 동시에 색인하여 목적에 따라 DB를 분리했다.

 

5. 느낀점

Kafka를 통해 생산자와 소비자를 분리함으로써, AI 서비스의 응답이 지연되거나 processor-service에 일시적인 장애가 발생하더라도 news-youtube-service는 아무런 영향 없이 계속해서 데이터를 수집할 수 있었다. 이러한 비동기 파이프라인은 시스템 전체의 회복탄력성과 확장성을 향상시키는 핵심적인 설계였다고 생각된다.