카프카 커넥트

카프카 커넥트(Kafka Connect)는 Apache Kafka의 데이터 통합 프레임워크로, 다양한 데이터 소스와 싱크를 Kafka로 연결하여 실시간으로 데이터를 전송하고 처리할 수 있도록 도와준다. 카프카 커넥트는 소스 시스템(데이터베이스, 파일 시스템, 메시지 큐 등)으로부터 데이터를 Kafka 토픽으로 가져오거나, Kafka 토픽에서 데이터를 추출하여 타겟 시스템(데이터베이스, Elasticsearch, HDFS 등)으로 보낼 수 있다. 카프카 커넥트는 확장 가능하고 유연한 아키텍처를 가지고 있으며, 단일 노드부터 클러스터 환경까지 지원한다.

[그림 1] 카프카 커넥트

 

커넥터와 태스크(Connector and Task)

카프카 커넥트에서 커넥터(Connector)는 데이터 통합 작업을 구성하는 기본 단위다. 각 커넥터는 특정 데이터 소스 또는 싱크와 상호 작용하도록 설계되며, 커넥터 설정을 통해 데이터 소스 및 싱크의 위치, 데이터 처리 방식 등을 지정할 수 있다.

태스크(Task)는 커넥터의 작업 단위를 분할하여 병렬로 실행하는 개념이다. 하나의 커넥터는 여러 개의 태스크로 분할될 수 있으며, 각 태스크는 데이터 소스 또는 싱크와 상호작용하여 데이터를 처리한다. 이를 통해 대량의 데이터를 효율적으로 처리할 수 있으며, 시스템의 처리 성능을 극대화할 수 있다.

예를 들어, 대용량 데이터베이스에서 데이터를 가져오는 소스 커넥터는 여러 태스크로 분할되어 병렬로 데이터를 가져와 Kafka로 전송할 수 있다.

[그림 2] 커넥터와 태스크

 

 

커넥트를 실행하는 방법(Running Kafka Connect)

카프카 커넥트는 다양한 방법으로 실행할 수 있으며, 단일 모드(standalone mode)와 분산 모드(distributed mode)를 제공한다. 사용자의 요구 사항에 따라 적절한 실행 방법을 선택할 수 있다.

단일 모드 커넥트(Standalone Mode)

단일 모드 커넥트는 주로 개발, 테스트 또는 소규모 배포 환경에서 사용된다. 단일 모드에서는 모든 작업이 하나의 프로세스에서 수행되며, 커넥터와 태스크는 로컬에서 실행된다. 이 모드는 설정이 간단하고 디버깅이 용이하지만, 확장성이나 내결함성(fault tolerance)이 부족하다.

단일 모드를 실행하려면, 설정 파일을 구성하고 connect-standalone 명령어를 사용하여 실행한다. 여기서 worker.properties는 커넥트 워커의 설정 파일이고, connector.properties는 커넥터의 설정 파일이다.

connect-standalone /path/to/worker.properties /path/to/connector.properties

 

 

[그림 3] 단일 모드 커넥트

 

분산 모드 커넥트(Distributed Mode)

분산 모드 커넥트는 프로덕션 환경에서 주로 사용되며, 고가용성 및 확장성을 제공한다. 분산 모드에서는 여러 노드에 커넥트 워커를 배포하여 클러스터를 구성하며, 각 노드는 태스크를 병렬로 실행하고, 작업을 공유한다. 이를 통해 시스템 장애 시에도 작업이 중단되지 않고, 새로운 워커 노드를 추가하여 성능을 확장할 수 있다.

분산 모드를 실행하려면, connect-distributed 명령어를 사용하며, 여러 노드에서 동시에 실행하여 클러스터를 구성한다.

분산 모드는 클러스터의 각 워커가 동일한 설정 파일을 사용하여 동일한 커넥터를 실행하도록 구성할 수 있으며, Kafka의 토픽을 통해 작업 상태를 관리한다.

connect-distributed /path/to/worker.properties

[그림 4] 분산 모드 커넥트

카프카 커넥트 REST API

GET	/connectors	모든 커넥터의 리스트를 조회
POST	/connectors	새로운 커넥터를 생성. 요청 본문에 커넥터 설정 정보가 포함되어야 한다.
GET	/connectors/{name}	특정 커넥터의 설정 정보를 조회
PUT	/connectors/{name}/config	특정 커넥터의 설정을 업데이트
DELETE	/connectors/{name}	특정 커넥터를 삭제
GET	/connectors/{name}/status	특정 커넥터의 현재 상태를 조회
POST	/connectors/{name}/restart	특정 커넥터를 재시작
GET	/connectors/{name}/tasks	특정 커넥터의 모든 태스크 목록을 조회
GET	/connectors/{name}/tasks/{task_id}/status	특정 태스크의 상태를 조회
POST	/connectors/{name}/tasks/{task_id}/restart	특정 태스크를 재시작
GET	/connector-plugins	사용 가능한 커넥터 플러그인의 목록을 조회
POST	/connector-plugins/{plugin_name}/config/validate	특정 플러그인의 설정이 유효한지 검증
GET	/connectors/{name}/topics	특정 커넥터가 사용하는 토픽의 목록을 조회
GET	/connectors/{name}/topics/{topic}	특정 커넥터의 특정 토픽 설정을 조회
PUT	/connectors/{name}/pause	특정 커넥터를 일시 중지(pause)
PUT	/connectors/{name}/resume	특정 커넥터의 일시 중지를 해제(resume)
GET	/cluster/status	커넥트 클러스터의 전체 상태를 조회
GET	/cluster/nodes	커넥트 클러스터에 포함된 노드의 목록을 조회
GET	/cluster/nodes/{node_id}	특정 노드의 정보를 조회
GET	/admin/loggers	커넥트 워커의 로깅 수준을 조회
PUT	/admin/loggers/{logger_name}	특정 로거의 로깅 수준을 업데이트

소스 커넥터(Source Connector)

소스 커넥터(Source Connector)는 외부 시스템에서 데이터를 가져와서 Kafka 토픽으로 전송하는 역할을 한다. 소스 커넥터는 다양한 데이터 소스(관계형 데이터베이스, NoSQL 데이터베이스, 파일 시스템, 클라우드 스토리지 등)와 통합되어, 데이터를 Kafka로 실시간 스트리밍한다. 일반적으로 소스 커넥터는 데이터를 폴링하거나 이벤트 기반으로 가져오며, Kafka에 적합한 형식으로 변환하여 전송한다.

예를 들어, MySQL 데이터베이스에서 데이터를 가져오는 소스 커넥터를 사용하면, MySQL 테이블의 데이터 변경 사항을 감지하여 Kafka 토픽으로 전달할 수 있다. 이 데이터를 활용하여 실시간 분석, 모니터링, 알림 시스템 등을 구축할 수 있다.

 

싱크 커넥터(Sink Connector)

싱크 커넥터(Sink Connector)는 Kafka 토픽에서 데이터를 가져와서 외부 시스템으로 전송하는 역할을 한다. 싱크 커넥터는 Kafka에서 수집된 데이터를 최종 목적지(데이터베이스, 파일 시스템, 검색 엔진, 클라우드 스토리지 등)로 보낸다. 이 과정에서 데이터를 필요에 따라 변환하거나 필터링할 수 있다.

예를 들어, Elasticsearch 싱크 커넥터를 사용하면, Kafka 토픽의 데이터를 Elasticsearch에 저장하여 검색 및 분석할 수 있다. 이러한 싱크 커넥터는 데이터를 다양한 시스템에 전송함으로써 데이터의 활용도를 극대화할 수 있다.