Kafka를 활용한 로컬 데이터파이프라인을 구성해보자

데이터 파이프라인을 설계할 때 가장 경계해야 할 것은 '목적 없는 도구의 맹목적인 도입'이다. 특히 로컬 환경에서 수집한 데이터를 클라우드로 전송하는 하이브리드 아키텍처에서는 각 컴포넌트의 리소스 효율성과 책임 분리(Separation of Concerns)가 시스템의 안정성을 결정한다.

본 포스팅에서는 로컬 환경에서 로봇 센서 데이터를 수집 및 가공하고, Apache Kafka를 거쳐 최적화된 포워더(Vector)를 통해 AWS S3로 적재하는 엔드투엔드(End-to-End) 데이터 파이프라인의 구축 과정과 기술적 의사결정을 정리한다.

1. 설계 목표 및 파이프라인 아키텍처
   본 아키텍처의 핵심은 데이터 형식에 따른 처리 채널의 분리와 경량화된 클라우드 전송(Egress)이다.

(캡처 권장: 파이썬 소스부터 Fluent Bit, Kafka, Vector를 거쳐 S3까지 이어지는 전체 구조도)

파이프라인은 다음과 같은 논리적 흐름을 갖는다.

Source: log_gen.py를 통한 두 가지 형태(JSON, Text)의 로그 파일 생성.

Ingestion: Fluent Bit를 활용한 파일 Tailing 및 라우팅.

Transform: Logstash를 거쳐 비정형 텍스트를 구조화(Grok 파싱).

Messaging: KRaft 모드 기반의 다중 토픽 Kafka 클러스터.

Egress & Storage: Vector를 통한 Kinesis Data Firehose 데이터 포워딩 및 최종 Amazon S3(JSON 형태) 적재.

2. Apache Kafka의 본질과 KRaft의 도입
   Kafka는 단순한 메시지 큐가 아니라 데이터를 지연 없이 흘려보내는 이벤트 스트리밍 플랫폼이다. 데이터를 생산하는 프로듀서(Producer)와 이를 소비하는 컨슈머(Consumer) 간의 의존성을 완벽히 분리한다.

(캡처 권장: factory-json-topic과 factory-text-topic, 그리고 내부 상태 관리용 토픽들이 보이는 화면)

이번 아키텍처에서는 기존 Zookeeper 기반의 구성을 탈피하고 KRaft(Kafka Raft) 모드를 도입했다.
과거 수백만 개의 파티션 환경에서 컨트롤러 병목을 유발하던 Zookeeper 의존성을 제거함으로써, 메타데이터를 내부 로컬 로그로 관리하여 아키텍처를 단순화하고 로컬 컨테이너 리소스 점유율을 최적화했다.

3. 수집 및 가공: Channel 분리와 Logstash Filter의 역할
   모든 데이터를 동일한 파이프라인으로 처리하는 것은 비효율적이다. 본 설계에서는 원본 데이터의 구조화 여부에 따라 라우팅을 이원화했다.

[ LAYER ]          [ COMPONENT ]          [ PROTOCOL / ACTION ]            [ DATA FORMAT ]
====================================================================================================
SOURCE       :     Python (log_gen.py)  ----( Write to File )---->  [ /sensor_logs/*.log ]
               |                                                                   |
               |                                                            ( Shared Volume )
               v                                                                   |
INGESTION    :     Fluent Bit (Agent)   <---( Tailing File  )----------------------+
               |          |
               |          | [ CHANNEL A: DIRECT BYPASS(json) ]
               |          +--------------------------------------------( Produce )--+
               |          |                                                         |
               |          | [ CHANNEL B: PROXY ROUTE(text) ]                        |
               |          +----( Forward )----> [ Logstash ] ----( Produce )--+     |
               |                                     |                        |     |
               v                                     v                        v     v
TRANSFORM    : (A: Bypass)  +--[ Grok    : Pattern Matching & Extract Fields   ]--+ |
               |            +--[ Mutate  : Data Type Conversion (Str->Float)   ]--+ |
               |            +--[ Tagging : Add Metadata for Conditional Alerts ]--+ |
               |                                                              |     |
MESSAGING    :     Kafka / MSK (Broker) <-------------------------------------+-----+
               |          |             ( Merge Structured & Processed Data )
               |          |
               |          +----( Topic A: factory-json-topic ) <--- [Channel A Result]
               |          +----( Topic B: factory-text-topic ) <--- [Channel B Result]
               v                                                               |
               v                                                               |
                                                                        kafka connect / ui
                                                                                |
                                                                                v               
VISUALIZE    :     OpenSearch (DB)      <---( Indexing & Search )---->  DASHBOARD (Kibana)
=================================================================================================

Channel A (직결): 이미 구조화된 JSON 데이터는 가공 레이어를 생략하고 즉시 factory-json-topic으로 전송하여 Latency를 최소화한다.

Channel B (가공): 비정형 Text 데이터는 Fluent Bit에서 HTTP(포트 5044)를 통해 Logstash로 전달된다.

이 과정에서 HTTP 통신 프로토콜 특성상 막대한 양의 네트워크 메타데이터(IP, 헤더, 버전 등)가 노이즈처럼 결합된다.
예시)

{
    "event":{
        "original":"[{\"date\":1777005819.42045,\"log\":\"\\\"[2026-04-24 13:43:38] ID=AI-FACTORY-001 |   TEMP:87.5 |   HUMI:42.1 |   STAT:RUNNING\\\"\"}]"},
        "host":{"ip":"172.19.0.3"},
        "user_agent":{"original":"Fluent-Bit"},
        "date":1.77700581942045E9,
        "url":{
            "path":"/",
            "domain":"logstash",
            "port":5044
        },
        "http":{
            "version":"HTTP/1.1",
            "method":"POST",
            "request":{
                "body":{
                    "bytes":"124"
                },
                "mime_type":"application/json"
            }
        },
        "log":"\"[2026-04-24 13:43:38] ID=AI-FACTORY-001 |   TEMP:87.5 |   HUMI:42.1 |   STAT:RUNNING\"","@version":"1",
        "@timestamp":"2026-04-24T04:43:40.412003029Z"
    }

이러한 오염된 데이터를 그대로 Kafka에 적재하면 하위 스토리지 비용 증가 및 쿼리 복잡도를 유발한다. 따라서 Logstash의 filter 블록 내에서 grok 패턴([%{TIMESTAMP_ISO8601}] ID=%{DATA}...)을 엄격하게 적용하여, 필요한 센서 지표만 추출한 정제된 구조화 데이터로 변환 후 factory-text-topic으로 발행했다.

4. 클라우드 전송 아키텍처: 왜 Kafka Connect 대신 Vector인가?
   로컬 Kafka에 모인 데이터를 AWS 클라우드(Firehose -> S3)로 보내기 위해 초기에는 Kafka Connect를 고려했으나, 엔지니어링 관점에서 명확한 한계가 존재했다.

Java 의존성이 높은 Kafka Connect는 무겁고, 플러그인 관리 및 API 기반 설정이 번거롭다. 이를 대체하기 위해 Datadog에서 개발한 Vector를 도입했다.

압도적인 경량화: Rust 기반으로 작성되어 메모리 사용량이 극히 적다. 로컬 개발 환경에서 매우 유리하다.

운영 복잡도 감소: 복잡한 커넥터 설치 없이, 단일 vector.yaml 파일 하나로 파이프라인(Kafka -> Vector -> Firehose) 구성이 완료된다.

5. 데이터 적재 확인 및 파이프라인 검증
   Vector를 통해 포워딩된 데이터는 AWS Kinesis Data Firehose의 버퍼링(크기/시간 조건)을 거쳐 최종적으로 S3 버킷에 JSON 형태로 안전하게 적재된다.

로컬 디렉토리에서 시작된 단 한 줄의 텍스트 로그가 경량 에이전트, 스트리밍 허브, 포워더를 거쳐 클라우드 스토리지에 무손실 적재되는 전체 라이프사이클을 검증했다.

6. 마치며
   로컬과 클라우드를 잇는 하이브리드 스트리밍 아키텍처가 안정적으로 구축되었다. 불필요한 시스템(Zookeeper, Kafka Connect)을 도려내고, 각 계층에 가장 효율적인 도구(KRaft, Vector)를 배치하여 파이프라인의 체급을 낮추면서도 처리량은 극대화했다.

S3에 적재된 JSON 데이터는 향후 Amazon Athena를 활용한 서버리스 SQL 쿼리 분석의 기반이 될 것이며, 실시간 탐지가 필요한 구간에는 Flink를 추가 배치하여 파이프라인을 확장해 나갈 예정이다.