데이터 엔지니어 · AI 엔지니어를 위한 핵심 요약
Kafka Connect는 Apache Kafka와 외부 시스템(데이터베이스, 파일 시스템, 메시지 큐 등) 간의 데이터를 대규모로, 신뢰성 높게 스트림하는 프레임워크입니다. 커넥터(Connector) 플러그인을 통해 소스(Source)에서 데이터를 읽어 Kafka 토픽에 쓰거나, 싱크(Sink)로 Kafka 토픽의 데이터를 외부 시스템에 전송합니다.
분산 모드(클러스터)로 실행하여 수백 개의 커넥터를 관리할 수 있습니다. Worker 노드를 추가하여 처리량을 선형적으로 확장.
오프셋 관리, 자동 장애 복구, Exactly-once 또는 At-least-once 전송 보장을 지원합니다.
REST API를 통해 커넥터의 생성, 모니터링, 중단을 쉽게 관리할 수 있습니다. 설정 파일 중심 운영.
데이터 파이프라인을 직접 코딩하는 대신 Kafka Connect를 사용하면 생산성, 유지보수성, 안정성을 크게 향상시킬 수 있습니다.
| 방식 | 장점 | 단점 | Kafka Connect 비교 |
|---|---|---|---|
| 직접 코딩 (Python/Java) | 완전한 제어권, 특수 로직 구현 가능 | 코드 복잡성 증가, 장애 처리/모니터링 부담, 확장성 구현 필요 | 프레임워크 제공 기능(오프셋, 재시도, 확장)으로 부담 감소 |
| ETL 도구 (Airflow 등) | 작업 스케줄링, 워크플로우 관리 강점 | 실시간 스트림 처리에는 부적합, 높은 오버헤드 | 실시간 스트리밍에 특화, 지속적인 데이터 흐름 |
| 메시지 큐 직접 연결 | 간단한 통신 가능 | 포맷 변환, 대규모 데이터 적재 문제, 모니터링 부족 | 표준화된 커넥터, 다양한 시스템 지원, 통합 모니터링 |
# Kafka 배포판에 Connect 포함 (Confluent Platform 또는 Apache Kafka) # Confluent Community 버전 다운로드 예시 wget https://downloads.confluent.io/confluent-community/7.x.x/confluent-community.tar.gz tar -xzf confluent-community.tar.gz cd confluent-7.x.x
# connect-distributed.properties 주요 설정 bootstrap.servers=localhost:9092 group.id=connect-cluster key.converter=org.apache.kafka.connect.storage.StringConverter value.converter=org.apache.kafka.connect.storage.StringConverter offset.storage.topic=connect-offsets config.storage.topic=connect-configs status.storage.topic=connect-statuses plugin.path=/usr/share/java/kafka-connect-plugins
# Kafka Connect 서버 시작 bin/connect-distributed connect-distributed.properties
간단한 파일 소스 커넥터를 통해 텍스트 파일의 내용을 Kafka 토픽에 전송합니다.
# file-source-config.json 생성
{
"name": "local-file-source",
"config": {
"connector.class": "FileStreamSource",
"tasks.max": "1",
"file": "/tmp/test-input.txt",
"topic": "file-source-topic"
}
}
# REST API로 커넥터 생성
curl -X POST -H "Content-Type: application/json" \
--data @file-source-config.json \
http://localhost:8083/connectors
# 입력 파일에 데이터 작성
echo "첫 번째 메시지" >> /tmp/test-input.txt
echo "두 번째 메시지" >> /tmp/test-input.txt
# Kafka 토픽 확인 (kafka-console-consumer 사용)
bin/kafka-console-consumer --bootstrap-server localhost:9092 \
--topic file-source-topic --from-beginning
| 설정 키 | 설명 | 예시 값 | 필수 여부 |
|---|---|---|---|
connector.class |
커넥터 구현 클래스 (플러그인 이름) | io.confluent.connect.jdbc.JdbcSourceConnector |
필수 |
tasks.max |
커넥터가 생성할 최대 태스크 수 (병렬성) | 2 |
필수 |
topics / topic |
(Sink) 대상 토픽 목록 / (Source) 출력 토픽 | orders, customers |
필수 |
key.converter |
메시지 키 변환 클래스 | org.apache.kafka.connect.json.JsonConverter |
필수 |
value.converter |
메시지 값 변환 클래스 | org.apache.kafka.connect.json.JsonConverter |
필수 |
batch.size (JDBC 등) |
한 번에 읽거나 쓰는 레코드 수 | 1000 |
선택 |
poll.interval.ms (Source) |
소스 시스템 조회 간격 (밀리초) | 5000 |
선택 |
errors.tolerance |
오류 허용 정책 (none, all) | all |
선택 |
원인: 플러그인 경로(plugin.path) 설정 오류, 클래스 없음.
해결: 플러그인 JAR이 올바른 디렉토리에 있고, worker 재시작 후 로그 확인.
원인: poll.interval.ms 너무 길거나, 태스크 수(tasks.max) 부족.
해결: 병렬성 증가, 배치 크기 조정, 모니터링으로 병목 현상 확인.
원인: offset.storage.topic 손실 또는 커넥터 재설정.
해결: 오프셋 토픽 보존 정책 확인, 커넥터 오프셋 초기화(재생) 절차 수행.
tasks.max를 조정하여 싱크/소스 작업을 분산./connectors/{name}/status)로 상태 확인.errors.tolerance=all로 일시적 오류 무시 가능 (주의 필요).