Kafka

Kafka가 파이프라인에서 하는 역할

Kafka로 데이터 파이프라인을 만든다고 할 때는 보통 두 가지 사용 사례를 말한다. 첫째는 Kafka가 두 끝점 중 하나인 파이프라인이다. Kafka에서 S3로 데이터를 옮기거나 MongoDB에서 Kafka로 가져오는 경우다. 둘째는 서로 다른 두 시스템 사이의 파이프라인에 Kafka를 중간자로 두는 것이다. Twitter에서 Elasticsearch로 데이터를 보낼 때 먼저 Twitter → Kafka, 다음 Kafka → Elasticsearch로 보내는 식이다.

Kafka Connect는 이 두 사용 사례가 LinkedIn 등 대규모 조직에서 실제로 쓰이는 것을 보고 0.9 버전에 추가되었다. Kafka를 데이터 파이프라인에 통합할 때 모든 조직이 똑같이 부딪히는 문제들을 각자 처음부터 풀게 두지 말고, Kafka 자체에 API로 넣어 해결하자는 발상이었다.

Kafka가 파이프라인에 제공하는 주된 가치는 파이프라인의 여러 단계 사이에서 매우 크고 신뢰할 수 있는 버퍼가 되어, 파이프라인 안의 데이터 생산자와 소비자를 효과적으로 분리(decoupling)하는 것이다. 여기에 신뢰성·보안·효율성이 더해져 대부분의 데이터 파이프라인에 잘 맞는다.

덧붙여, 데이터 통합 문제에 부딪혔을 때 눈앞의 두 끝점만 보지 말고 더 큰 그림을 보라는 조언이 중요하다. 단기적인 통합에만 집중하다 보면 복잡하고 유지 비용이 높은 데이터 통합 난장판에 도달하게 된다.

적시성, 신뢰성, 처리량

**적시성(Timeliness).** 어떤 시스템은 하루에 한 번 대량으로 데이터가 도착하기를 기대하고, 어떤 시스템은 생성 후 몇 밀리초 안에 도착하기를 기대한다. 좋은 데이터 통합 시스템은 파이프라인마다 다른 적시성 요구를 지원하고, 비즈니스 요구가 바뀔 때 시간표를 옮기기도 쉬워야 한다. Kafka는 확장 가능하고 신뢰할 수 있는 저장소를 갖춘 스트리밍 데이터 플랫폼이므로 준실시간 파이프라인부터 시간 단위 배치까지 모두 지원한다. 프로듀서는 필요한 만큼 자주 또는 드물게 쓸 수 있고, 컨슈머는 도착하는 대로 읽거나 한 시간마다 접속해 쌓인 이벤트를 배치로 읽을 수 있다. Kafka를 생산자와 소비자의 시간 민감도 요구를 분리하는 거대한 버퍼로 보면 이해가 쉽다. 소비 속도가 전적으로 컨슈머에 의해 결정되므로 백프레셔 적용도 자연스럽다(필요할 때 ack를 지연시키는 형태로 Kafka 자체가 프로듀서에 백프레셔를 건다).

**신뢰성(Reliability).** 단일 장애점을 피하고 온갖 장애로부터 빠르고 자동적으로 복구되어야 한다. 파이프라인은 비즈니스 크리티컬 시스템에 데이터를 공급하는 통로인 경우가 많아 몇 초의 장애도 큰 지장이 된다. 전달 보장도 중요한 고려사항이다. 대개는 at-least-once가 요구되고(모든 이벤트가 목적지에 도달하되 재시도로 중복이 생길 수 있음), 종종 exactly-once까지 요구된다. Kafka는 자체적으로 at-least-once를 제공하고, 트랜잭션 모델이나 고유 키를 지원하는 외부 저장소와 결합하면 exactly-once를 제공한다. Connect API는 오프셋을 다룰 때 외부 시스템과 연동할 수 있는 API를 제공해 커넥터가 종단 간 exactly-once 파이프라인을 만들기 쉽게 해 준다.

**높고 변화하는 처리량.** Kafka가 생산자와 소비자 사이의 버퍼가 되므로 소비 처리량을 생산 처리량에 맞출 필요가 없다. 프로듀서 처리량이 컨슈머를 앞지르면 컨슈머가 따라잡을 때까지 Kafka에 데이터가 쌓이므로 복잡한 백프레셔 메커니즘을 구현할 필요도 없다. 컨슈머나 프로듀서를 독립적으로 추가해 파이프라인 양쪽을 각각 확장할 수 있다. 또한 Connect API는 단순한 스케일아웃이 아니라 작업의 병렬화에 초점을 두어, 소스와 싱크가 작업을 여러 실행 스레드로 나누어 한 대에서도 가용 CPU를 활용하게 한다. Kafka는 여러 압축 방식도 지원해 처리량 요구가 커질 때 네트워크와 스토리지 자원 사용을 조절할 수 있게 한다.

데이터 포맷, 변환(ETL vs ELT), 보안, 실패 처리

**데이터 포맷.** 파이프라인에서 가장 중요한 고려사항 중 하나는 서로 다른 데이터 포맷과 타입을 조율하는 것이다. XML과 관계형 데이터를 Kafka에 싣고, Kafka 안에서는 Avro를 쓰고, Elasticsearch에 쓸 때는 JSON으로, HDFS에 쓸 때는 Parquet으로, S3에 쓸 때는 CSV로 바꿔야 할 수 있다. Kafka 자체와 Connect API는 데이터 포맷에 대해 완전히 중립적이다. Kafka Connect는 데이터 타입과 스키마를 포함하는 자체 인메모리 객체를 갖지만, 플러그인 방식의 컨버터를 통해 이 레코드를 어떤 포맷으로든 저장할 수 있게 한다. 즉 Kafka에서 어떤 데이터 포맷을 쓰든 커넥터 선택이 제한되지 않는다.

많은 소스와 싱크에는 스키마가 있으므로, 소스에서 데이터와 함께 스키마를 읽어 저장하고 호환성 검증이나 싱크 DB의 스키마 갱신에 쓸 수 있다. MySQL에서 Hive로 가는 파이프라인이 고전적인 예다. 누군가 MySQL에 컬럼을 추가하면, 좋은 파이프라인은 새 데이터를 적재하면서 Hive에도 그 컬럼이 추가되도록 한다. 또한 Kafka에서 외부 시스템으로 쓸 때 그 포맷을 결정하는 것은 싱크 커넥터의 책임이며, HDFS 커넥터가 Avro와 Parquet 중에서 고르게 하듯 이를 플러그인화한 커넥터도 있다.

포맷뿐 아니라 소스·싱크의 동작 방식 차이도 다룰 수 있어야 한다. Syslog는 데이터를 밀어내는 소스이고 관계형 DB는 프레임워크가 데이터를 당겨와야 하는 소스다. HDFS는 append-only라 쓰기만 가능하지만 대부분의 시스템은 추가와 갱신을 모두 허용한다.

**변환: ETL과 ELT.** ETL(Extract-Transform-Load)은 데이터가 지나가는 동안 파이프라인이 데이터를 수정하는 방식이다. 저장하고 수정하고 다시 저장할 필요가 없어 시간과 스토리지를 아끼는 것처럼 보이지만, 그 부담이 파이프라인 자체로 옮겨 갈 뿐일 수 있다. 가장 큰 단점은 파이프라인에서 일어난 변환이 하류에서 데이터를 처리하려는 사람들의 손을 묶는다는 것이다. MongoDB와 MySQL 사이 파이프라인을 만든 사람이 어떤 이벤트를 걸러내거나 필드를 제거하기로 했다면, MySQL의 데이터를 쓰는 모든 사용자와 애플리케이션은 부분적인 데이터만 보게 된다. 빠진 필드가 필요해지면 파이프라인을 다시 만들고 과거 데이터를 재처리해야 한다(그 데이터가 남아 있다면).

ELT(Extract-Load-Transform)는 파이프라인이 최소한의 변환(주로 데이터 타입 변환)만 수행해 타깃에 도착하는 데이터가 원본과 최대한 비슷하게 만드는 방식이다. 고충실도(high-fidelity) 파이프라인 또는 데이터 레이크 아키텍처라고도 한다. 타깃 시스템이 '원시 데이터'를 모으고 필요한 처리는 모두 타깃에서 한다. 장점은 타깃 사용자가 모든 데이터에 접근할 수 있어 유연성이 극대화되고, 모든 처리가 파이프라인과 애플리케이션으로 쪼개지지 않고 한 시스템에 모여 있어 트러블슈팅이 쉽다는 것이다. 단점은 변환이 타깃 시스템의 CPU와 스토리지를 소모한다는 점이며, 그 시스템이 비싼 경우 연산을 밖으로 밀어내려는 동기가 강하게 생긴다.

**보안.** 파이프라인에서의 주요 관심사는 세 가지다. 파이프를 지나는 데이터가 암호화되는가(주로 데이터센터 경계를 넘는 파이프라인에서 중요), 누가 파이프라인을 수정할 수 있는가, 접근 제어된 위치를 읽고 쓸 때 제대로 인증할 수 있는가. Kafka는 소스에서 Kafka로, Kafka에서 싱크로 가는 구간의 전송 중 암호화를 지원하고, SASL을 통한 인증과 인가를 지원한다. 따라서 민감한 정보를 담은 토픽이 권한 없는 사람에 의해 보안 수준이 낮은 시스템으로 흘러가지 않게 할 수 있다. 또 접근을 추적하는 감사 로그를 제공하며, 약간의 추가 코딩으로 각 토픽의 이벤트가 어디서 왔고 누가 수정했는지까지 추적해 레코드별 계보를 제공할 수 있다.

**실패 처리.** 모든 데이터가 언제나 완벽하리라 가정하는 것은 위험하다. 잘못된 레코드가 애초에 파이프라인에 들어오지 못하게 할 수 있는가, 파싱할 수 없는 레코드에서 복구할 수 있는가, 잘못된 레코드를 (사람이) 고쳐서 재처리할 수 있는가, 나쁜 이벤트가 정상 이벤트와 똑같이 생겼고 며칠 뒤에야 문제를 발견하면 어떻게 할 것인가를 미리 계획해야 한다. Kafka는 모든 이벤트를 오랜 기간 저장하므로 필요할 때 과거로 돌아가 오류를 복구할 수 있다.

결합도와 민첩성을 해치는 세 가지

파이프라인의 가장 중요한 목표 중 하나는 데이터 소스와 타깃을 분리하는 것이다. 의도치 않은 결합은 여러 경로로 생긴다.

**애드혹 파이프라인.** 연결하려는 애플리케이션 쌍마다 맞춤 파이프라인을 만드는 회사들이 있다. 로그를 Elasticsearch로 넣는 데 Logstash, HDFS로 넣는 데 Flume, Oracle에서 HDFS로 옮기는 데 GoldenGate, MySQL과 XML을 Oracle로 넣는 데 Informatica를 쓰는 식이다. 파이프라인이 특정 끝점에 강하게 결합되고 배포·유지·모니터링에 큰 노력이 드는 통합 지점의 난장판이 된다. 새로운 시스템을 도입할 때마다 파이프라인을 추가로 만들어야 하므로 신기술 도입 비용이 올라가고 혁신이 저해된다.

**메타데이터 손실.** 파이프라인이 스키마 메타데이터를 보존하지 않고 스키마 진화를 허용하지 않으면 소스의 생산 소프트웨어와 목적지의 소비 소프트웨어가 강하게 결합된다. 스키마 정보가 없으면 양쪽 모두 데이터를 어떻게 파싱하고 해석할지에 대한 정보를 코드에 담아야 한다. Oracle에서 HDFS로 데이터가 흐르는데 DBA가 스키마 정보 보존이나 진화 지원 없이 Oracle에 필드를 추가하면, HDFS에서 데이터를 읽는 모든 앱이 깨지거나 모든 개발자가 동시에 애플리케이션을 업그레이드해야 한다. 둘 다 민첩하지 않다. 파이프라인이 스키마 진화를 지원하면 각 팀이 자기 속도로 애플리케이션을 수정할 수 있다.

**과도한 처리.** 데이터 파이프라인에는 어느 정도의 처리가 본질적으로 따른다. 서로 다른 데이터 포맷이 어울리는 시스템들 사이로 데이터를 옮기기 때문이다. 그러나 처리가 지나치면 파이프라인을 만들 때 내린 결정에 모든 하류 시스템이 묶인다. 어떤 필드를 보존할지, 데이터를 어떻게 집계할지 같은 결정들이다. 하류 애플리케이션의 요구가 바뀔 때마다 파이프라인을 계속 고쳐야 하므로 민첩하지도, 효율적이지도, 안전하지도 않다. 더 민첩한 방식은 가능한 한 많은 원시 데이터를 보존하고 하류 앱이 스스로 처리와 집계를 결정하게 하는 것이다.

Connect를 쓸 것인가, 프로듀서/컨슈머를 쓸 것인가

Kafka 클라이언트는 자기 애플리케이션 안에 임베드되는 클라이언트다. 연결하려는 애플리케이션의 코드를 수정할 수 있고, Kafka로 데이터를 밀어 넣거나 Kafka에서 데이터를 당겨오려 할 때 Kafka 클라이언트를 쓴다.

반면 Connect는 직접 작성하지 않았고 코드를 수정할 수 없거나 수정하고 싶지 않은 데이터 저장소에 Kafka를 연결할 때 쓴다. 외부 저장소에서 Kafka로 데이터를 당겨오거나 Kafka에서 외부 저장소로 밀어내는 용도다. 이미 커넥터가 존재하는 저장소라면 개발자가 아닌 사람도 커넥터를 설정만 해서 Connect를 쓸 수 있다.

커넥터가 아직 없는 저장소에 Kafka를 연결해야 한다면 Kafka 클라이언트로 앱을 짜는 것과 Connect API를 쓰는 것 중에 고를 수 있는데, Connect가 권장된다. 설정 관리, 오프셋 저장, 병렬화, 오류 처리, 다양한 데이터 타입 지원, 표준 관리용 REST API를 기본으로 제공하기 때문이다. Kafka와 저장소를 잇는 작은 앱을 짜는 일은 간단해 보이지만, 데이터 타입과 설정에 관한 자잘한 문제가 많아 결코 사소하지 않다.

Connect 실행: 워커, 분산 모드와 스탠드얼론 모드

Kafka Connect는 Apache Kafka에 포함되어 있어 따로 설치할 필요가 없다. 다만 프로덕션에서 대량의 데이터를 옮기거나 커넥터를 많이 돌릴 계획이라면 별도 서버에서 Connect를 실행해야 한다. 모든 머신에 Apache Kafka를 설치하고, 일부 서버에서는 브로커를, 다른 서버에서는 Connect를 시작하면 된다.

Connect 워커를 시작하는 방법은 브로커와 비슷하게 프로퍼티 파일을 주고 시작 스크립트를 호출하는 것이다.

bin/connect-distributed.sh config/connect-distributed.properties

주요 워커 설정은 다음과 같다.

- `bootstrap.servers`: Connect가 함께 동작할 Kafka 브로커 목록. 클러스터의 모든 브로커를 적을 필요는 없지만 최소 3개를 지정하는 것이 권장된다. - `group.id`: 같은 그룹 ID를 가진 워커들이 하나의 Connect 클러스터를 이룬다. 클러스터에서 시작된 커넥터는 어느 워커에서든 실행될 수 있고 그 태스크들도 마찬가지다. - `key.converter`와 `value.converter`: Kafka에 저장할 메시지의 키와 값에 대한 컨버터를 지정한다. 기본값은 Apache Kafka에 포함된 JSONConverter를 쓰는 JSON 포맷이며, Confluent Schema Registry의 AvroConverter로 설정할 수도 있다.

컨버터별 추가 설정도 있다. JSON 메시지는 스키마를 포함할 수도, 포함하지 않을 수도 있어서 `key.converter.schema.enable`을 true/false로 설정한다. 값 컨버터에는 `value.converter.schema.enable`을 쓴다. Avro 메시지도 스키마를 담지만 Schema Registry 위치를 `key.converter.schema.registry.url`과 `value.converter.schema.registry.url`로 지정해야 한다.

커넥터는 보통 Kafka Connect의 REST API로 설정하고 모니터링하며, `rest.host.name`과 `rest.port`로 그 포트를 지정한다. 워커가 뜬 뒤에는 REST API로 확인할 수 있다.

curl http://localhost:8083/
{"version":"0.10.1.0-SNAPSHOT","commit":"561f45d747cd2a8c"}

curl http://localhost:8083/connector-plugins
[{"class":"org.apache.kafka.connect.file.FileStreamSourceConnector"},
 {"class":"org.apache.kafka.connect.file.FileStreamSinkConnector"}]

한편 Connect에는 스탠드얼론 모드도 있다. `bin/connect-distributed.sh` 대신 `bin/connect-standalone.sh`를 실행하고, REST API 대신 커넥터 설정 파일을 명령줄로 전달할 수도 있다. 이 모드에서는 모든 커넥터와 태스크가 하나의 스탠드얼론 워커에서 돈다. 개발과 트러블슈팅에 편하고, 커넥터와 태스크가 특정 머신에서 실행되어야 하는 경우(예: 특정 포트를 리스닝하는 syslog 커넥터)에도 쓰인다.

커넥터 예제: 파일 소스/싱크와 JDBC → Elasticsearch

파일 소스 커넥터는 Kafka 설정 파일을 읽어 토픽으로 흘려보내는 식으로 시험할 수 있다. REST API에 커넥터 이름과 설정 맵을 담은 JSON을 POST한다.

echo '{"name":"load-kafka-config", "config":{"connector.class":"FileStreamSource","file":"config/server.properties","topic":"kafka-config-topic"}}' | curl -X POST -d @- http://localhost:8083/connectors --header "content-Type:application/json"

토픽을 콘솔 컨슈머로 읽으면 server.properties가 한 줄씩 JSON으로 변환되어 들어와 있는 것을 볼 수 있다. 기본적으로 JSON 컨버터는 레코드마다 스키마를 함께 넣는데, 이 경우 스키마는 문자열 타입의 payload 컬럼 하나뿐이라 매우 단순하다.

반대 방향의 싱크 커넥터도 같은 방식으로 만든다.

echo '{"name":"dump-kafka-config", "config":{"connector.class":"FileStreamSink","file":"copy-of-server-properties","topics":"kafka-config-topic"}}' | curl -X POST -d @- http://localhost:8083/connectors --header "content-Type:application/json"

소스와 달라진 점에 주목할 만하다. 클래스가 FileStreamSource에서 FileStreamSink로 바뀌었고, file 속성이 이제 소스가 아니라 목적지 파일을 가리키며, `topic`(단수)이 아니라 `topics`(복수)를 지정한다. 싱크는 여러 토픽을 한 파일에 쓸 수 있지만 소스는 하나의 토픽에만 쓸 수 있기 때문이다.

커넥터 삭제는 다음과 같다.

curl -X DELETE http://localhost:8083/connectors/dump-kafka-config

삭제 후 워커 로그를 보면 다른 커넥터들이 태스크를 재시작하는 것을 볼 수 있다. 커넥터가 사라진 뒤 남은 태스크들을 워커 사이에 리밸런싱해 부하를 고르게 맞추기 위해서다.

더 실용적인 예로 MySQL 테이블을 Kafka 토픽으로 스트리밍한 뒤 Elasticsearch로 적재할 수 있다. JDBC 소스 커넥터는 connection.url, mode(예: timestamp), table.whitelist, timestamp.column.name, topic.prefix 등을 설정한다. 어떤 설정이 가능한지는 문서를 봐도 되지만 REST API로도 알아낼 수 있다. 빈 설정을 `connector-plugins/{커넥터}/config/validate`에 PUT하면 사용 가능한 모든 설정의 JSON 정의가 돌아온다.

Elasticsearch 싱크 커넥터에서는 connection.url이 Elasticsearch 서버 주소이고, Kafka의 각 토픽이 기본적으로 토픽과 같은 이름의 별도 인덱스가 된다. 토픽 안의 데이터 타입을 type.name으로 지정하며, 예제의 MySQL 테이블에 기본 키가 없어 Kafka 이벤트의 키가 null이었기 때문에 `key.ignore=true`를 주어 토픽 이름·파티션 ID·오프셋을 각 이벤트의 키로 쓰도록 알려 준다. 실제로 색인된 문서의 _id가 `mysql.login+0+1` 형태인 것을 확인할 수 있다. 이후 MySQL 테이블에 행을 추가하면 자동으로 Kafka 토픽과 Elasticsearch 인덱스에 반영된다.

Connect 내부: 커넥터, 태스크, 워커, 컨버터, 오프셋 관리

**커넥터와 태스크.** 커넥터 플러그인은 커넥터 API를 구현하며 두 부분으로 나뉜다.

커넥터는 세 가지를 책임진다. 이 커넥터에 대해 태스크를 몇 개 실행할지 결정하고, 데이터 복사 작업을 태스크들에 어떻게 나눌지 정하고, 워커로부터 태스크용 설정을 받아 전달하는 것이다. 예컨대 JDBC 소스 커넥터는 데이터베이스에 접속해 복사할 테이블을 발견하고, 그에 따라 필요한 태스크 수를 `max.tasks` 설정값과 테이블 수 중 작은 쪽으로 결정한다. 태스크 수가 정해지면 커넥터 설정(예: connection.url)과 각 태스크가 복사할 테이블 목록을 조합해 태스크별 설정을 생성한다. `taskConfigs()` 메서드가 맵의 리스트, 즉 실행할 태스크마다의 설정을 반환한다. 워커는 태스크를 시작하고 각각에 고유한 설정을 주어 서로 다른 테이블 부분집합을 복사하게 한다. REST API로 커넥터를 시작하면 그 커넥터가 어느 노드에서 시작될지, 그것이 만든 태스크들이 어느 노드에서 실행될지는 정해져 있지 않다.

태스크는 실제로 데이터를 Kafka에 넣고 빼는 일을 한다. 모든 태스크는 워커로부터 컨텍스트를 받아 초기화된다. 소스 컨텍스트에는 소스 레코드의 오프셋을 저장할 수 있게 해 주는 객체가 들어 있다(파일 커넥터에서는 파일 내 위치, JDBC 소스에서는 테이블의 기본 키 ID 등). 싱크 커넥터의 컨텍스트에는 Kafka에서 받는 레코드를 제어하는 메서드들이 들어 있어 백프레셔 적용, 재시도, exactly-once 전달을 위한 외부 오프셋 저장 등에 쓰인다. 초기화 후 태스크는 커넥터가 만든 설정을 담은 Properties 객체와 함께 시작된다. 시작된 소스 태스크는 외부 시스템을 폴링해 레코드 목록을 반환하고 워커가 이를 브로커로 보낸다. 싱크 태스크는 워커를 통해 레코드를 받아 외부 시스템에 쓴다.

**워커.** 워커 프로세스는 커넥터와 태스크를 실행하는 '컨테이너' 프로세스다. 커넥터와 그 설정을 정의하는 HTTP 요청을 처리하고, 커넥터 설정을 저장하고, 커넥터와 태스크를 시작하며 적절한 설정을 전달한다. 워커 프로세스가 멈추거나 크래시하면 Connect 클러스터의 다른 워커들이 (Kafka 컨슈머 프로토콜의 하트비트로) 이를 인지하고 그 워커에서 돌던 커넥터와 태스크를 남은 워커들에 재할당한다. 새 워커가 합류하면 다른 워커들이 이를 알아채고 커넥터나 태스크를 할당해 부하가 고르게 분산되도록 한다. 워커는 소스·싱크 커넥터 양쪽의 오프셋 자동 커밋과 태스크가 오류를 던졌을 때의 재시도도 담당한다.

워커를 이해하는 가장 좋은 방법은 역할 분담을 보는 것이다. 커넥터와 태스크는 데이터 통합에서 '데이터를 옮기는' 부분을, 워커는 REST API·설정 관리·신뢰성·고가용성·확장·부하 분산을 담당한다. 이 관심사 분리가 Connect API를 고전적인 컨슈머/프로듀서 API 대신 쓰는 가장 큰 이점이다. Kafka에서 데이터를 읽어 DB에 넣는 코드는 하루 이틀이면 짜지만, 설정·오류·REST API·모니터링·배포·확장·장애 처리까지 제대로 하려면 몇 달이 걸린다는 것을 경험 있는 개발자들은 안다.

**컨버터와 데이터 모델.** Connect API는 데이터 객체와 그것을 기술하는 스키마를 포함하는 데이터 API를 갖는다. JDBC 소스는 DB에서 컬럼을 읽고 반환된 컬럼의 데이터 타입을 바탕으로 Connect Schema 객체를 구성한 뒤, 그 스키마로 레코드의 모든 필드를 담은 Struct를 만든다. 컬럼마다 이름과 값을 저장하는 식이다. 모든 소스 커넥터가 이처럼 소스 시스템의 이벤트를 읽어 Schema와 Value 쌍을 만든다. 싱크 커넥터는 그 반대로 Schema와 Value 쌍을 받아 Schema로 값을 해석해 타깃 시스템에 넣는다.

그렇다면 워커는 이 객체들을 Kafka에 어떻게 저장할까? 여기서 컨버터가 등장한다. 사용자가 워커(또는 커넥터)를 설정할 때 Kafka에 데이터를 저장하는 데 쓸 컨버터를 고른다. 책 집필 시점의 선택지는 Avro, JSON, 문자열이다. JSON 컨버터는 결과 레코드에 스키마를 포함할지 말지 설정할 수 있어 구조화 데이터와 반구조화 데이터를 모두 지원한다. 커넥터가 데이터 API 레코드를 워커에 반환하면 워커는 설정된 컨버터로 그것을 Avro 객체, JSON 객체, 문자열 중 하나로 변환해 Kafka에 저장한다. 싱크 커넥터는 정반대 과정을 거친다. 덕분에 커넥터 구현과 무관하게 Kafka에 저장되는 데이터 타입을 자유롭게 고를 수 있다.

**오프셋 관리.** 워커가 커넥터에게 제공하는 편의 서비스 중 하나가 오프셋 관리다. 커넥터는 이미 처리한 데이터를 알아야 하고, Kafka가 제공하는 API로 그 정보를 유지할 수 있다.

소스 커넥터의 경우, 커넥터가 워커에 반환하는 레코드에는 논리적 파티션과 논리적 오프셋이 담긴다. 이는 Kafka 파티션이나 Kafka 오프셋이 아니라 소스 시스템에서 필요한 형태의 파티션과 오프셋이다. 파일 소스에서 파티션은 파일이고 오프셋은 파일 안의 줄 번호나 문자 위치일 수 있으며, JDBC 소스에서 파티션은 테이블, 오프셋은 테이블 레코드의 ID일 수 있다. 소스 커넥터를 작성할 때 가장 중요한 설계 결정 중 하나가 소스 시스템의 데이터를 어떻게 분할하고 오프셋을 어떻게 추적할지인데, 이것이 커넥터가 달성할 수 있는 병렬성 수준과 at-least-once인지 exactly-once인지를 좌우한다. 소스 커넥터가 소스 파티션·오프셋이 붙은 레코드 목록을 반환하면 워커는 그것을 브로커로 보내고, 브로커가 성공적으로 ack하면 워커가 보낸 레코드들의 오프셋을 저장한다. 저장 메커니즘은 플러그인 방식이며 보통 Kafka 토픽이다. 그래서 재시작이나 크래시 후 가장 최근에 저장된 오프셋부터 이벤트 처리를 이어갈 수 있다.

싱크 커넥터는 반대이지만 비슷하다. 이미 토픽·파티션·오프셋 식별자를 가진 Kafka 레코드를 읽고, 커넥터의 `put()` 메서드를 호출해 그 레코드를 목적지 시스템에 저장하게 한다. 커넥터가 성공을 보고하면 워커는 커넥터에 넘겼던 오프셋을 통상적인 컨슈머 커밋 메서드로 Kafka에 커밋한다.

Kafka Connect의 대안

**다른 데이터 저장소의 인제스트 프레임워크.** 어떤 조직은 데이터 아키텍처를 Hadoop이나 Elasticsearch 중심으로 구성한다. 그런 시스템에는 자체 데이터 인제스트 도구가 있다. Hadoop에는 Flume, Elasticsearch에는 Logstash나 Fluentd다. Kafka가 아키텍처의 핵심이고 다수의 소스·싱크를 연결하는 것이 목표라면 Connect API를 권장한다. 반대로 Hadoop 중심 또는 Elastic 중심 시스템을 만들고 있고 Kafka가 그 시스템의 여러 입력 중 하나일 뿐이라면 Flume이나 Logstash를 쓰는 편이 합리적이다.

**GUI 기반 ETL 도구.** Informatica 같은 오래된 시스템, Talend·Pentaho 같은 오픈소스 대안, Apache NiFi·StreamSets 같은 새로운 대안 모두 Apache Kafka를 소스이자 목적지로 지원한다. 이미 그런 시스템을 쓰고 있다면(예: 모든 것을 Pentaho로 처리 중이라면) Kafka만을 위해 또 다른 데이터 통합 시스템을 추가하고 싶지 않을 것이고, GUI 기반으로 ETL 파이프라인을 만드는 접근을 선호한다면 역시 합리적이다. 다만 이런 시스템은 대개 복잡한 워크플로를 위해 만들어졌기 때문에, 단지 Kafka로 데이터를 넣고 빼기만 하려는 목적에는 다소 무겁고 번거로운 해법이 된다. 데이터 통합은 어떤 조건에서도 메시지를 충실히 전달하는 데 집중해야 하는데 대부분의 ETL 도구는 불필요한 복잡성을 더한다는 것이 이 책의 관점이다. Kafka는 데이터 통합(Connect), 애플리케이션 통합(프로듀서/컨슈머), 스트림 처리를 모두 다룰 수 있는 플랫폼이며, 데이터 저장소 통합만 하는 ETL 도구의 유효한 대체재가 될 수 있다.

**스트림 처리 프레임워크.** 거의 모든 스트림 처리 프레임워크가 Kafka에서 이벤트를 읽어 다른 시스템에 쓰는 기능을 포함한다. 목적지 시스템이 지원되고 이미 그 프레임워크로 Kafka의 이벤트를 처리할 계획이라면 데이터 통합에도 같은 프레임워크를 쓰는 것이 합리적이다. 처리된 이벤트를 Kafka에 저장할 필요 없이 바로 다른 시스템에 쓰면 되므로 워크플로에서 한 단계를 줄이는 경우가 많다. 단점은 메시지 유실이나 손상 같은 문제를 트러블슈팅하기 더 어려워질 수 있다는 점이다.