내가 지금 참고하고 있는 책은 실전 카프카 개발부터 운영까지 (저자: 고승범) 인데
최근에 다 읽었다고 하고 뭔가 허전한 거 같아서 들춰보니... 카프카 커넥트 부분을 정리하지 않았다.
([도서 회고록] "실전 카프카 개발부터 운영까지"를 읽고)
(너란 녀석 좀... 이젠 징글징글하다... )
생각을 해 보니 해당 부분을 읽다가 12장을 보다가 순간 빨리 끝내고 싶다는 마음에
11장을 다 읽었다고 스스로 착각하고 바로 후루룩! 넘겨버렸다.
이미 다 읽었다고 했지만 해당 부분을 이번 글에서 추가적으로 정리할 것이다.
목표
1. Kafka Connect에 대해 이해한다.
2. Kafka Connect가 실제로 어떻게 동작하는지 실습을 통해 학습한다.
1. Kafka Connect의 정의와 장점 알아보기
1.1) Kafka Connect의 정의
Apache Kafka 공식문서에서는 Kafka Connect를 다음과 같이 정의하고 있다.
Kafka Connect는 Apache Kafka와 다른 시스템 간에 확장 가능하고 안정적인 데이터 스트리밍을 위한 도구입니다.
출처 : https://kafka.apache.org/43/kafka-connect/overview/
Confluent은 Kafak Connect 다음과 같이 정의하고 있다.
Kafka Connect는 Apache Kafka®의 무료 오픈 소스 구성 요소로, 데이터베이스, 키-값 저장소, 검색 인덱스 및 파일 시스템 간의 간단한 데이터 통합을 위한 중앙 집중식 데이터 허브 역할을 합니다.
출처 : https://docs.confluent.io/platform/current/connect/index.html#what-is-kafka-connect
즉, Apache Kafka와 다른 시스템간에 데이터 통합을 도와주는 도구! 라고 이해하면 된다.
1.2) Kafka Connect의 장점
- 데이터 중심 파이프라인 (Data-centric pipeline) : Connect는 유의미한 데이터 추상화를 활용하여 Kafka로 데이터를 가져오거나(Pull) 밀어넣습니다(Push).
- 재사용성 및 기능 확장성 (Reusability and extensibility) : Connect는 기존 커넥터를 그대로 활용하거나 요구사항에 맞게 커스텀(확장)할 수 있으며, 이를 통해 프로덕션(상용) 환경 배포까지 소요되는 시간을 단축합니다.
- 유연성 및 스케일아웃 (Flexibility and scalability) : Connect는 단일 노드(Standalone) 환경에서 스트리밍 및 배치 기반 시스템과 함께 실행되거나, 전사적 규모의 분산 서비스(Distributed)로 스케일아웃할 수 있습니다.
뭐... 다 좋다고 하겠지.
카프카 커넥트를 사용하면 무엇보다도 다양한 소스 데이터를 카프카 안으로 처리하기도 쉽고 밖으로 보내기도 쉽다.
Kafka Connect를 위하 사용할 수 있는 말 그대로 수백가지의 다른 connectors들이 있다.
다음은 가장 인기가 많은 것들이다.
- RDBMS (Oracle, SQL Server, Db2, Postgres, MySQL)
- Cloud object stores (Amazon S3, Azure Blob Storage, Google Cloud Storage)
- Message queues (ActiveMQ, IBM MQ, RabbitMQ)
- NoSQL and document stores (Elasticsearch, MongoDB, Cassandra)
- Cloud data warehouses (Snowflake, Google BigQuery, Amazon Redshift)
가장 강력한 부분은 다음 부분인 것 같다.
...
카프카 커넥트(Kafka Connect)의 가장 큰 장점은 별도의 프로그래밍(코딩)이 필요 없다는 점입니다.
이 도구는 전적으로 설정(Configuration) 기반으로 동작하므로, 개발자뿐만 아니라 폭넓은 직군의 사용자들이 쉽게 접근하고 활용할 수 있습니다.
또한, 데이터의 수집(Ingest) 및 추출(Egress) 기능 외에도 데이터가 파이프라인을 통과하는 과정에서 가벼운 변환(Lightweight transformation) 작업을 수행할 수 있습니다.
...
출처 : https://developer.confluent.io/courses/kafka-connect/intro/
코딩 없이 설정만으로 사용이 가능하다고? 좀 치네!
2. Kafka Connect의 핵심 개념과 구성 및 내부 동작
Apache Kafka 의 공식문서보다는 Confluent 의 공식 문서가 설명이 더 보기 좋았다.
Confluent공식 문서에서는 개념에 대해 깔끔하게 정리해줘서 이해하는게 훨씬 좋았다.
Kafka Connect 개념 이해하기
카프카 커넥트에는 다음이 존재한다.
- 커넥터(Connectors) : 데이터 스트리밍을 조정하고 작업을 관리하는 상위 수준의 추상화 개념
- 작업 혹은 태스크(Tasks) : Kafka로 또는 Kafka에서 데이터를 복사하는 방법 구현
- 워커(Workers) : 커넥터와 작업을 실행하는 실행 중인 프로세스
- 변환기(Converters) : Connect 와 데이터를 송수신하는 시스템 간의 데이터 변환에 사용되는 코드
- 변환(Transforms) : 커넥터에서 생성되거나 커넥터로 전송되는 각 메시지를 변경하는 간단한 로직
- 데드 레터 큐 (Dead Letter Queue) : 처리에 실패한 레코드를 따로 보내두는 카프카 토픽. 커넥터가 아니라 토픽이다.
주의할 점은 DLQ가 싱크 커넥터에서만 동작한다는 것이다.
소스 커넥터는 외부에서 카프카로 밀어 넣는 방향이라 실패한 레코드를 흘려보낼 지점 자체가 없다.
이 글의 실습은 CDC(소스)라서 DLQ는 등장하지 않는다.
구성은 다음을 보면 이해가 쉽다.
여기서 재미있는 건 ingest 라는 단어를 쓴다. (음식을) 섭취하다 라는 뜻 말고도 (데이터를) 가져오다 라는 뜻도 있다고 한다.
다음을 번역해서 읽어보자.
카프카 커넥트는 어떻게 동작하나?
카프카 커넥트(Kafka Connect) 프레임워크를 사용하면 전체 데이터베이스를 수집하거나 모든 애플리케이션 서버의 메트릭을 카프카 토픽(Kafka topics)으로 가져와, 짧은 지연 시간(Low Latency)으로 스트림 처리에 활용할 수 있습니다.
예를 들어 익스포트(Export) 커넥터는 카프카 토픽의 데이터를 엘라스틱서치(Elasticsearch)와 같은 보조 인덱스나 오프라인 분석을 위한 하둡(Hadoop) 같은 배치 시스템으로 전달할 수 있습니다.
카프카 커넥트는 단일 머신에서 작업을 실행하는 독립형(Standalone) 프로세스(예: 로그 수집)로 배포하거나,
조직 전체를 지원하는 분산형(Distributed), 확장 가능하며 내결함성(Fault-tolerant)을 갖춘 서비스로 배포할 수 있습니다.
카프카 커넥트는 진입 장벽과 운영 오버헤드가 낮습니다. 개발 및 테스트를 위한 독립형 환경으로 작게 시작한 다음, 대규모 조직의 데이터 파이프라인을 지원하는 완전한 운영(Production) 환경으로 규모를 확장할 수 있습니다.
출처 : https://docs.confluent.io/platform/current/connect/index.html#how-kafka-connect-works
좀 더 자세히 들여다보기 위해 다음과 같이 그림을 그렸다.
여기서 커넥트, 커넥터 이렇게 헷갈리는데 connect를 연결해주는 도구들을 가리키는 일반적인 개념이라 본다면
Connector는 실제 연결 도구(즉, 플러그인) 로 이해했다.
- Connect = 프레임워크, 그리고 워커들이 모인 클러스터
- Connector = 그 위에서 도는 플러그인 (Debezium MySQL 등)
여기서 플러그인이 뭐라고? connect의 구현체라고 이해하자.
양방향에 커넥트가 존재하는데, 동일한 두 커넥트를 서로 구분하기 위해 소스 방향에 있는 커넥트를 소스 커넥트,
싱크 방향에 있는 커넥트를 싱크 커넥트라고 한다.
소스 커넥트와 싱크 커넥트 내부를 보면 워커1, 워커2 그리고 워커3이 있다.
현재 그림에서는 각각의 커텍트에서 총 3개의 워커(인스턴스)를 실행한 분산 모드 소스 커넥트와 분산 모드 싱크 커넥트를 나타냈다.
카프카 커넥트는 단일모드와 분산모드가 있다고 한다.
두 모드의 진짜 차이는 상태를 어디에 저장하느냐다. 여기서 상태란 (1) 어디까지 읽었는지(오프셋), (2) 커넥터 설정, (3) 커넥터/태스크의 현재 상태 세 가지를 말한다.
- 단일 모드(Standalone): 오프셋을 워커 서버의 로컬 파일(offset.storage.file.filename)에 저장한다. 커넥터 설정은 .properties 파일로 프로세스 기동 시 주입한다.
- 분산 모드(Distributed): 세 가지 상태를 각각 connect-offsets, connect-configs, connect-status라는 카프카 내부 토픽에 저장한다. 커넥터는 REST API로 등록한다.
상태가 카프카에 있으니 워커가 죽어도 다른 워커가 그 태스크를 이어받을 수 있다.
이게 분산 모드가 장애 복구가 되는 이유다. 단일 모드는 상태가 그 서버 로컬 디스크에만 있으니 프로세스가 죽으면 그걸로 끝이다.
단일 모드가 구식이라는 뜻은 아니다.
엣지 노드에서 로그 파일 하나 긁어 올리는 것처럼 워커가 물리적으로 한 대일 수밖에 없는 상황에서는 지금도 정상적인 선택지다.
| 구분 | Standalone (단일 모드) | Distributed (분산 모드) |
| 오프셋 저장 위치 | 워커 로컬 파일 | connect-offsets 토픽 |
| 커넥터 설정 저장 | .properties 파일 | connect-configs 토픽 |
| 커넥터 등록 방식 | 기동 시 파일 주입 | REST API (POST) |
| 설정 변경 | 프로세스 재시작 필요 | REST 호출로 즉시 반영 |
| 워커 장애 시 | 중단 | 다른 워커로 태스크 재분배 |
| 적합한 상황 | 엣지 수집, 단일 노드 확정 | 운영 환경, CDC 파이프라인 |
분산 모드의 소스 커넥터와 싱크 커넥터 안을 자세히 보면 커넥터와 태스크가 존재한다.
아까 정리를 따르면 태스크란 Kafka로 또는 Kafka에서 데이터를 복사하는 방법을 구현한다고 했다.
각 태스크들은 메시지들을 소스에서 카프카로 혹은 카프카에서 싱크로 이동시키는 역할을 한다.
즉, 소스에서 소스 커넥터를 거쳐 kafka로 아니면 kafka에서 싱크 커넥터를 거쳐 싱크로 데이터를 이동시키는 역할을 한다.
다음 그림을 보자.
그림을 자세히 보면 Stream안에 파티션이 있다.
이 파티션들을 Connector 내부의 Task1, 2가 각각 파티션 1,2와 파티션 3,4를 통해 데이터를 받아온다.
여기서 헷갈리는 부분이 잇는데 Stream 이고 Partition이 나와서 으흠? 카프카가 또뭐 나온건가? 싶은데
우리가 읽어와야 할 거대한 외부 소스(DB 테이블, 거대한 로그 파일 등)를 병렬 처리가 가능하도록 논리적/물리적으로 잘라놓은 청크(Chunk)들의 집합을 표현한 것일 뿐이다.
커넥터(Connector)가 병렬 처리를 위해 1개의 거대한 원본 데이터를 스스로 여러 개의 '논리적 작업 단위(Partition)'로 쪼개서 읽어온다고 이해하자.
이 부분이 너무 추상적이고 와닿지가 않아서 예시를 Gemini에게 질문했다. 궁금하면 다음을 열어보자.
Spring Boot와 카프카 커넥트를 분리하는 구조 (CDC 패턴)
가장 권장되는 일반적인 아키텍처입니다. Spring Boot 애플리케이션은 카프카 커넥트의 존재를 전혀 알 필요가 없습니다.
Spring Boot는 평소처럼 JPA를 사용해 비즈니스 로직을 처리하고 DB에 데이터를 저장합니다. 그러면 DB와 연결된 카프카 커넥트(독립된 프로세스)가 DB의 트랜잭션 로그를 읽어 문서의 정의대로 데이터를 변환합니다.
PMS 결제 도메인을 예로 들면 다음과 같이 매핑됩니다.
- Spring Boot의 역할: Payment 엔티티를 생성하고 JPA save()를 호출하여 MySQL에 INSERT 쿼리를 발생시킵니다.
- 카프카 커넥트의 추상화 (문서 내용의 실체화):
- 파티션 (Partition): MySQL의 payment_table 자체, 혹은 DB 트랜잭션 로그(binlog)의 특정 구간.
- 레코드의 Key: 결제 내역의 PK (예: payment_id = 12345)
- 레코드의 Value: 결제 전체 데이터 (예: {"amount": 50000, "status": "APPROVED", "method": "CARD"})
- 순서가 지정된 시퀀스: DB에 커밋된 시간 순서대로 이벤트가 발행됨을 보장.
이 구조에서 백엔드 개발자가 Spring Boot 서버를 위해 해야 할 일은, 카프카 커넥트와 연동하기 위한 코드를 짜는 것이 아니라 DB의 테이블과 PK(Key)를 잘 설계하여 커넥터가 상태를 고유하게 식별할 수 있도록 만드는 것입니다.
정리하면 이렇다. Source Partition은 커넥터가 원본 데이터를 논리적으로 쪼갠 단위다.
그런데 이걸 나누는 1차 목적은 병렬 처리가 아니라 "어디까지 읽었는지"를 기록하기 위해서다.
커넥터는 레코드를 뱉을 때 sourcePartition과 sourceOffset을 항상 같이 붙인다.
sourcePartition: {"file": "mysql-bin.000003"}
sourceOffset: {"pos": 154}
이 쌍이 connect-offsets 토픽에 저장된다. 워커가 죽었다 살아나면 이 값을 읽어서 "binlog 000003 파일의 154 위치부터 다시"를 결정한다. 파티션이 여러 개면 병렬로 읽을 수 있다는 건 그 위에 얹히는 부수 효과다.
한 가지 더! partition이라는 단어가 소스와 싱크에서 완전히 다른 걸 가리킨다.
- 소스에서의 partition: 커넥터가 임의로 정의하는 논리적 단위 (파일명, 테이블명, binlog 좌표)
- 싱크에서의 partition: 진짜 카프카 토픽 파티션. 태스크 분배가 컨슈머 그룹의 파티션 할당을 그대로 따라간다
헷갈렸던 이유가 이거였다. 같은 단어인데 가리키는 대상이 다르다.
3. Kafka Connect Use Case 와 실습하기(스탠드얼론)
3.1) Kafka Connect Use Case
Confluent에서 소개하는 Use Case는 다음과 같다.
A) 스트리밍 파이프라인(Streaming Pipeline)
카프카 커넥트는 실시간 데이터 흐름을 처리하는 데 사용할 수 있는데, 전통적인 데이터베이스를 소스로도 가능하다.
B) 카프카로부터 데이터저장소에 쓰기(Writing to Datastore from Kafka)
특정한 로그 데이터를 DB에 적재할 때 활용할 수 있다.
C) 구버전에서 신버전으로 프로세스 개선(Evolve Processing from Old Systems to New)
CDC(Change Data Capture)를 통해서 데이터를 Kafka Cluster에 적재하면 다른 Application이 이를 활용하는 형식이다.
D) 시스템을 실시간으로 만들기(Make Systems Real Time)
원본 DB에 데이터가 쓰이는 순간(INSERT/UPDATE), 카프카 커넥트가 DB의 트랜잭션 로그(예: MySQL의 binlog, Oracle의 redo log)를 감지하여 그 변경 사항을 즉시 카프카 이벤트로 발행한다.
필자는 단일 환경에서는 파일 복사를 해보았다. 처음에는 둘 다 하려고 했는데, 어짜피 이거 AI써서 딸깔 설정 다하는거면
애써서 이걸 외울 필요가 없고, 아키텍트의 마음가짐으로 개념이 무엇인지, 어떻게 돌아가는지만 알고자! 단일모드만 실습하기로 했다. (아 귀찮아~)
어떤 connector가 있는지 모른다면 Kafka Connectors for Confluent Platform에서 찾아보면 되는데, 우리가 직접 만들 필요 없이 필요한 connector는 대부분 다 있다고 한다. configuration만 잘 설정하면 된다고 했는데 정말 그런지도 볼 것이다.
물론 AI를 열심히 활용할 것이다!
3.2) 단일모드 실습하기
필자는 미니 PC로 실습을 하고 있는데 미니 PC에서 컨테이너 현황을 파악했다.
# 모든 컨테이너(종료 상태 포함)의 이름과 포트 매핑 정보를 탭(Tab)으로 구분하여 추출
$ docker ps -a --format '{{.Names}}\t{{.Ports}}'
schema-registry
kafka-kraft-3
kafka-kraft-1 9092/tcp, 0.0.0.0:19092->19092/tcp, [::]:19092->19092/tcp
kafka-kraft-2 9092/tcp, 0.0.0.0:29092->29092/tcp, [::]:29092->29092/tcp
grafana
prometheus
#혹은 다르게
$ docker ps --format 'table {{.Names}}\t{{.Status}}\t{{.Ports}}'
현재 뭐 띄운건 없으니 이상하게 겹쳐서 작동하지 않을 일은 없을 것이다.
큰 그림은 다음과 같다.
여기서 다음과 같이 직접 kafka를 다 받아서 할 수도 있지만 docker써서 하겠다.
직접 다운받는 kafka
cd ~
curl -O https://downloads.apache.org/kafka/4.3.1/kafka_2.13-4.3.1.tgz
tar -xzf kafka_2.13-4.3.1.tgz
cd kafka_2.13-4.3.1
구성도
~/kafka-connect/
├── docker-compose.yml
├── test.txt ← 이미 있는 파일 (소스)
├── test.sink.txt ← 실행 후 자동 생성 (싱크 결과)
├── connect.offsets ← 실행 후 자동 생성 (오프셋)
├── config/
│ ├── connect-standalone.properties
│ ├── connect-file-source.properties
│ └── connect-file-sink.properties
└── plugins/
└── file-connector/
└── connect-file-4.x.x.jar
디렉토리 준비
# kafka-connect 디렉토리로 이동
cd ~/kafka-connect
# 디렉토리 연쇄 생성
mkdir -p config plugins/file-connector
# 기존 test.txt 확인
cat test.txt
# 컨테이너가 이 디렉토리에 파일을 써야 함
chmod 777 ~/kafka-connect
플러그인 jar 꺼내기
# 공식 카프카 도커 이미지 내부에 있는 File Connector JAR 파일을
# 호스트 머신의 ~/kafka-connect/plugins/file-connector 디렉토리로 추출(복사)한 뒤,
# 작업을 수행한 임시 컨테이너는 즉시 삭제하는 명령어
docker run --rm -v ~/kafka-connect/plugins/file-connector:/out apache/kafka:latest \
sh -c 'cp /opt/kafka/libs/connect-file-*.jar /out/ && ls -l /out/'
docker-compose.yml 설정
# << 'EOF' 부터 마지막 EOF 까지의 내용을 ~/kafka-connect/docker-compose.yml 파일에 기록합니다.
cat > ~/kafka-connect/docker-compose.yml << 'EOF'
services:
# --------------------------------------------------------
# 1. 카프카 브로커 인스턴스 (KRaft 모드)
# --------------------------------------------------------
kclab-broker:
image: apache/kafka:latest # 아파치 카프카 공식 최신 이미지를 사용합니다.
container_name: kclab-broker # 컨테이너의 식별자입니다.
hostname: kclab-broker # 컨테이너 내부 네트워크에서 사용할 호스트명입니다.
networks: [kclab-net] # 아래 정의된 사용자 정의 브리지 네트워크에 연결합니다.
ports:
- "39092:39092" # [호스트포트]:[컨테이너포트] - 외부(호스트 OS) 클라이언트가 접근할 포트입니다.
environment:
# [아키텍처 포인트] 주키퍼 의존성을 제거한 KRaft 모드 설정입니다.
KAFKA_NODE_ID: 1 # KRaft 클러스터 내에서 이 브로커를 식별하는 고유 ID입니다.
KAFKA_PROCESS_ROLES: broker,controller # 하나의 노드가 브로커(데이터 처리)와 컨트롤러(메타데이터 관리) 역할을 동시에 수행합니다.
# [네트워크 포인트] 네트워크 인터페이스 분리 (컨트롤러, 내부 브로커 간, 외부 클라이언트용)
KAFKA_LISTENERS: INTERNAL://:9092,CONTROLLER://:9093,EXTERNAL://:39092
# 각 리스너가 클라이언트에게 반환할 접속 주소(Advertised)입니다. 외부 클라이언트는 localhost:39092로 접근하게 됩니다.
KAFKA_ADVERTISED_LISTENERS: INTERNAL://kclab-broker:9092,EXTERNAL://localhost:39092
KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: INTERNAL:PLAINTEXT,CONTROLLER:PLAINTEXT,EXTERNAL:PLAINTEXT # 모든 통신은 평문(PLAINTEXT)을 사용합니다.
KAFKA_INTER_BROKER_LISTENER_NAME: INTERNAL # 브로커 간 데이터 복제 등에 사용할 내부 통신망 지정입니다.
KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER # 컨트롤러 노드 간 합의(Quorum)에 사용할 리스너 지정입니다.
KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kclab-broker:9093 # 메타데이터 뗏목(Raft) 합의 알고리즘에 참여할 투표 노드 목록입니다.
# [인프라 포인트] 단일 노드 테스트 환경을 위한 복제본 수 강제 조정
KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1 # 컨슈머 오프셋 저장용 내부 토픽의 복제본 수 (기본값은 3이나 노드가 1개이므로 1로 낮춤)
KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1 # 트랜잭션 상태 관리 로그 복제본 수
KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1 # 트랜잭션 커밋을 위해 동기화되어야 할 최소 복제본 수
KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0 # 새로운 컨슈머 그룹 합류 시 대기 시간 (빠른 테스트를 위해 0초로 설정)
KAFKA_AUTO_CREATE_TOPICS_ENABLE: "false" # 존재하지 않는 토픽에 메시지 인입 시 자동 생성되는 것을 방지합니다. (명시적 관리를 위해 false 권장)
# --------------------------------------------------------
# 2. 카프카 커넥트 인스턴스 (Standalone 모드)
# --------------------------------------------------------
kclab-connect:
image: apache/kafka:latest # 브로커와 동일한 이미지를 사용하되, 실행 진입점을 변경하여 커넥트로 활용합니다.
container_name: kclab-connect
hostname: kclab-connect
networks: [kclab-net]
depends_on: [kclab-broker] # 브로커가 먼저 실행된 후 커넥트가 실행되도록 순서를 제어합니다.
ports:
- "8084:8083" # 호스트 8084 포트로 커넥트 REST API(기본 8083 포트)에 접근할 수 있게 엽니다.
volumes:
- .:/data # 호스트의 현재 디렉토리를 컨테이너의 /data 경로로 양방향 마운트합니다.
entrypoint: ["/opt/kafka/bin/connect-standalone.sh"] # 컨테이너 시작 시 실행할 바이너리를 단일(Standalone) 커넥트 스크립트로 덮어씌웁니다.
command: # entrypoint의 인자(Arguments)로 3개의 설정 파일을 전달합니다.
- /data/config/connect-standalone.properties # 커넥트 워커 자체의 구동 설정 (브로커 주소, 오프셋 저장 방식 등)
- /data/config/connect-file-source.properties # Source 커넥터(파일 읽기) 생성 설정
- /data/config/connect-file-sink.properties # Sink 커넥터(파일 쓰기) 생성 설정
restart: on-failure # 프로세스 충돌 등으로 비정상 종료 시에만 컨테이너를 재시작합니다.
# --------------------------------------------------------
# 3. 네트워크 정의
# --------------------------------------------------------
networks:
kclab-net: # 브로커와 커넥트가 IP가 아닌 컨테이너 이름(DNS)으로 통신할 수 있게 해주는 브리지 네트워크입니다.
name: kclab-net
EOF
워커파일 3개 설정
cat > ~/kafka-connect/config/connect-standalone.properties << 'EOF'
# --- 모든 워커 공통 ---
bootstrap.servers=kclab-broker:9092
key.converter=org.apache.kafka.connect.storage.StringConverter
value.converter=org.apache.kafka.connect.storage.StringConverter
plugin.path=/data/plugins
# --- 스탠드얼론 전용 ---
offset.storage.file.filename=/data/connect.offsets
offset.flush.interval.ms=10000
# --- REST API ---
listeners=http://0.0.0.0:8083
EOF
소스커넥터와 싱크커넥터 설정
cat > ~/kafka-connect/config/connect-file-source.properties << 'EOF'
name=local-file-source
connector.class=FileStreamSource
tasks.max=1
file=/data/test.txt
topic=connect-test
EOF
cat > ~/kafka-connect/config/connect-file-sink.properties << 'EOF'
name=local-file-sink
connector.class=FileStreamSink
tasks.max=1
file=/data/test.sink.txt
topics=connect-test
EOF
브로커 기동 + 토픽 생성
cd ~/kafka-connect
docker compose up -d kclab-broker
# 준비 대기
docker logs -f kclab-broker
# "Kafka Server started" 나오면 Ctrl+C
# 토픽 생성
docker exec kclab-broker /opt/kafka/bin/kafka-topics.sh \
--bootstrap-server localhost:9092 \
--create --topic connect-test \
--partitions 1 --replication-factor 1
# 확인
docker exec kclab-broker /opt/kafka/bin/kafka-topics.sh \
--bootstrap-server localhost:9092 --describe --topic connect-test
Connect 기동
docker compose up -d kclab-connect
docker logs -f kclab-connect
순서대로 잘 실행한 후 test.sink.txt를 보면 뭐... 우리가 입력한 값이 동일하게 찍혀잇는것을 확인할 수 있다.
그리고 또 기존의 test.txt 를 수정하면 추가가 된다.
또한 내부에 offset도 확인이 가능하다.
kafka 어휴... 내가 뭐가 궁금하다고 이걸 또 공부하겠다고 들쑤셔서 이렇게 머리만 아프나...
후회는 아니지만 뭔가 일을 벌리니 심심할 틈이 없다.
하지만 늘 근육 시바견 짤처럼 오른쪽에서 왼쪽으로 변한 나의 개발 실력을 상상하며... 공부하자!
최근 Gemini의 성능이 심상치 않다(안좋은 쪽으로)
이번 글은 사실 직접 찾아가면서 하긴 했는데 실습은 claude에게 A-Z 구성을 부탁했다.
최근의 트렌드는 역시... coding은 맞길 수 있으나 이를 지휘하는 인간의 역할이 중요한 것 같다.
아무것도 모르는 나같은 문외한도 할 수 있게 해주니 말이다.
출처 :
카프카 공식문서 - Apache Kafka Connect : Overview
Confluent 공식 문서 - Kafka Connect for Confluent Platform
도서 - 실전 카프카 개발부터 운영까지 중 11장 : Kafka Connect
COURSE: KAFKA CONNECT 101 - Introduction to Kafka Connect
출처 : 카프카 커넥트를 데이터 파이프라인으로 사용하는 이유? kafka-sink-connector 오픈소스 언빡싱!
'programming language > Kakfa' 카테고리의 다른 글
| [Kafka 활용] Kafka 지식 부록 1 - Kafka MirrorMaker2, Kafka Stretched Cluster(feat. 토스증권 사례 + Uber 의 KIP-405) (2) | 2026.07.27 |
|---|---|
| [Kafka 활용] Kafka 심화 4 - 스키마 레지스트리와 실습하기 (0) | 2026.07.24 |
| [Kafka 활용] Kafka 심화 3 - 보안 적용 : Kafka SSL, SASL, ACL (0) | 2026.07.17 |
| [Kafka 활용] Kafka 심화 2 - 안정적인 운영을 위한 지식 한꼬집 + 모니터링 도구 한꼬집 (0) | 2026.07.06 |
| [Kafka 활용] Kafka 심화 1 - Uber System Design와 Kafka 사용 (0) | 2026.07.04 |