[Apache Kafka] connect, connector

Kafka connect는 Kafka와 다른 데이터 시스템(ex DB, elastic search 등) 간에 확장 가능하고 안정적으로 데이터를 스트리밍하기 위한 도구입니다. 데이터 소스와 Kafka 토픽 사이의 데이터 이동을 쉽게 관리하고 스트리밍 데이터 파이프라인을 구축하는 데 도움이 됩니다.

connector는 데이터를 Kafka로 가져오거나 Kafka에서 가져와 외부 시스템으로 보내는 플러그인입니다. Kafka connect는 내장된 커넥터 외에도 다양한 커넥터를 제공하며, 사용자 정의 커넥터를 개발하여 특정 데이터 소스와 통합할 수 있습니다.

 

아래의 명령어로 connect를 실행할 수 있습니다.

.\connect-distributed.bat ..\..\config\connect-distributed.properties

 

connector는 GET 요청을 http://localhost:8083/connectors 로 보내면 조회가 가능합니다.

현재는 connector가 없기 때문에 빈 배열이 응답되고 있습니다.

connector 조회

 

connector-plugin은 GET 요청을 http://localhost:8083/connector-plugins 로 보내면 조회가 가능합니다.

현재는 내장된 커넥터만 응답이 되고 있기 때문에 mysql connector를 설치를 해줘야 합니다.

connector plugin 조회

 

connector는 source connector와 sink connector 두 가지 종류가 있습니다.

source connector는 외부 시스템에서 데이터를 읽어와 Kafka 토픽으로 전송하고, sink connector는 Kafka 토픽에서 데이터를 가져와 외부 시스템으로 보냅니다.

 

저는 kafka topic에 있는 데이터를 mysql로 보내기 위해서 sink connector를 생성해보도록 하겠습니다.

sink connector를 생성하기 위해서는 아래와 같은 json 데이터를 POST 요청으로 http://localhost:8083/connectors 로 보냅니다.

{
    "name":"event-sink-connect",
    "config":{
        "connector.class":"io.confluent.connect.jdbc.JdbcSinkConnector",
        "connection.url":"jdbc:mariadb://localhost:3306/데이터베이스명",
        "connection.user":"아이디",
        "connection.password":"비밀번호",
        "auto.create":"false",
        "auto.evolve":"false",
        "delete.enabled":"false",
        "tasks.max":"1",
        "topics":"토픽명"
    }
}

현재 sink connectror 를 생성하려고 하면 아래와 같이 JdbcSinkConnector 클래스가 없어서 실패합니다.

connector 생성 요청

 

JDBC Connector를 설치 하기 위해 confluent사이트에서 설치를 해줍니다.

해당 사이트에서 Download 버튼을 클릭하면 zip파일을 다운받을수 있습니다.

JDBC connector 다운로드 사이트

 

다운받은 zip파일을 적절한 경로로 옮긴 후 압축을 풀고 kafka connect를 실행할때 입력했던 connect-distributed.properties 파일의 에디터로 열어서 plugin.path에 압축을 푼 폴더의 lib경로를 추가해줍니다.

그리고 다시 connector를 생성해보면 정상적으로 생성이 됩니다.

sink connector 생성

 

connector를 조회해보면 요청한 connector가 생성되었음을 확인할 수 있습니다.

connector 조회

 

이제 connector를 생성할때 전달한 topic명에 메세지가 들어가면 sink connector에 의해서 mysql에 있는 topic명과 동일한 테이블에 데이터가 쌓이게 됩니다.

 

 

참고자료

https://docs.confluent.io/kafka-connectors/jdbc/current/sink-connector/overview.html#install-the-connector-using-c-hub

https://velog.io/@ksh9409255/Kafka-Connect

https://cjw-awdsd.tistory.com/53