카프카를 다른 데이터 저장소와 연결하기 위한 API와 런타임. 코드를 짤 수 없거나 바꿀 수 없는 저장소에 카프카를 붙여야 할 때 쓰고, 사용자가 하는 일은 설정 파일을 작성하는 것뿐이다.
워커가 떠맡는 관심사
커넥터와 태스크는 데이터 이동만 맡고 워커가 나머지 전부를 맡는다. REST API와 설정 관리, 신뢰성과 고가용성, 규모 확장성과 부하 분산, 오프셋을 내부 토픽에 자동 커밋하는 일과 에러 시 재시도가 전부 워커 몫이다.
이 관심사의 분리가 고전적인 프로듀서와 컨슈머 대신 커넥트 API를 쓰는 주된 이점이다. 같은 것을 직접 만들면 설정 관리, 에러 처리, REST API, 모니터링, 배포, 규모 확장, 장애 대응을 전부 짜야 하고 제대로 동작하게 만드는 데만 몇 달이 걸린다. 워커가 죽으면 다른 워커들이 감지해서 커넥터와 태스크를 재할당하고, 새 워커가 들어오면 부하가 다시 균형을 잡는다.
워커, 커넥터, 태스크
이름이 비슷한 셋이 역할을 나눠 가진다. 카프카 커넥트는 커넥터를 실행해주는 프로세스이고, 커넥터는 데이터를 어디에서 어디로 복사할지 관리한다. 실제 실행 단위는 셋으로 나뉘는데, 워커는 커넥터와 태스크를 실행하는 컨테이너 프로세스이고, 커넥터는 태스크를 몇 개 띄울지와 데이터 복사 작업을 어떻게 분할할지 결정하며, 태스크는 실제로 데이터를 카프카에 넣거나 가져온다.
데이터 API와 컨버터
커넥트 API에는 데이터 객체와 그 구조를 나타내는 스키마를 함께 다루는 데이터 API가 들어 있다. JDBC 소스 커넥터가 DB의 열을 읽으면 리턴된 타입에 따라 ConnectSchema 객체를 만든다. 모든 소스 커넥터가 원본의 이벤트를 읽어 스키마와 값의 순서쌍을 만들고, 싱크 커넥터는 정확히 반대 작업을 한다.
그 객체를 카프카에 어떤 형식으로 쓸지 정하는 것이 컨버터다. JSON, Avro, Protobuf, 스키마 있는 JSON, 문자열, 바이트 배열을 지원한다. 컨버터 덕분에 커넥터는 데이터 형식과 무관해진다. 사용 가능한 컨버터만 있으면 어떤 커넥터든 레코드 형식에 상관없이 쓸 수 있다.
원본 시스템의 오프셋
오프셋 관리도 워커가 커넥터에 제공하는 편리한 기능이다. 여기서 말하는 오프셋은 카프카의 것이 아니라 원본 시스템의 것이다. 파일 소스라면 파일 자체가 파티션 역할을 하고 파일 안의 줄이나 문자 위치가 오프셋 역할을 하며, JDBC라면 테이블이 파티션이고 레코드의 ID나 타임스탬프가 오프셋이다. 소스 커넥터를 개발할 때 가장 중요한 것이 이 결정인데, 원본 데이터를 어떻게 분할하고 오프셋을 어떻게 추적하는지가 커넥터의 병렬성 수준과 전달 의미구조를 정하기 때문이다. 순서도 중요하다. 워커는 브로커가 레코드를 성공적으로 썼다는 응답을 받은 뒤에야 그 오프셋을 저장한다. 그래서 재시작이나 크래시 이후에도 마지막 저장 지점부터 이어갈 수 있다.
SMT로 하는 단위 레코드 변환
상태 없는 단위 레코드 변환은 SMT(Single Message Transformation)가 맡고 대부분 코드 없이 설정만으로 된다.
Cast는 필드의 데이터 타입을 바꾼다MaskField는 특정 필드의 내용물을 null로 채운다. 민감한 정보나 개인 식별 정보를 제거할 때 쓴다Filter는 조건에 부합하는 메시지를 제외하거나 포함한다Flatten은 중첩된 자료 구조를 편다InsertField와ReplaceField는 필드를 추가하거나 삭제하거나 이름을 바꾼다RegexRouter는 정규식과 교체할 문자열로 목적지 토픽의 이름을 바꾼다TimestampConverter는 필드의 시간 형식을 바꾼다TimestampRouter는 메시지의 타임스탬프를 기준으로 토픽을 바꾼다. 싱크 커넥터에서 타임스탬프 기준으로 나뉜 테이블 파티션에 넣을 때 쓴다
조인이나 집계처럼 상태가 필요한 변환은 SMT로 할 수 없다. 카프카 스트림즈의 영역이다.
워커 실행과 설정
프로덕션에서는 브로커와 별도의 서버에서 워커를 띄운다.
$ bin/connect-distributed.sh config/connect-distributed.propertiesbootstrap.servers는 함께 작동할 브로커 목록으로 최소 3개 이상이 권장되고, group.id가 같은 워커들이 하나의 커넥트 클러스터를 이룬다. plugin.path는 커넥터와 그 의존성을 찾을 디렉토리이고, key.converter와 value.converter의 기본값은 JSONConverter다. 커넥터 설정과 모니터링은 rest.host.name과 rest.port로 여는 REST API로 하는 것이 보통이다.
로그스태시와 ETL 툴
카프카를 중심으로 쓰고 있다면 커넥트가 낫고, 엘라스틱서치를 중심으로 쓰고 있다면 로그스태시가 낫다. 하둡의 플룸, Fluentd도 같은 판단 기준이다.
인포매티카, Talend, Pentaho, Apache NiFi 같은 GUI 기반 ETL 툴은 복잡한 워크플로를 상정하고 만들어져서, 단순히 데이터를 옮기는 것이 목적이면 무겁고 복잡하다.