kafka - connector 구성요소

구성 요소 ( Connector, Converter, Transform )

Kafka Connect는 여러 컴포넌트로 구성된 

플러그형 아키텍처를 바탕으로 유연한 데이터 통합 파이프라인을 제공합니다. 

주요 컴포넌트로는 데이터 소스 및 싱크와 상호작용하는 커넥터(Connectors), 

데이터 직렬화/역직렬화를 담당하는 컨버터(Converters), 

데이터 변환을 수행하는 변환기(Transforms)가 있습니다.

 

Connector

Kafka Connect 파이프라인의 주요 구성 요소는

데이터가 어디에서 복사되어야 하는지를 정의하는 커넥터 인스턴스입니다.

 

커넥터 인스턴스에 의해 구현되거나 사용되는 모든 클래스는 해당 커넥터 플러그인에 정의되어 있고,

이러한 플러그인은 코드 작성을 하지 않고 다운로드, 설치 및 사용할 수 있는 재사용 가능한 구성 요소입니다.

커넥터 플러그인 자체는 카프카에서 읽거나 쓰지 않습니다.

 

예시)

• Debezium MySQL 소스 커넥터는 MySQL bin log를 사용하여 데이터베이스에서 이벤트를 읽고 Kafka Connect에 스트리밍
• Elasticsearch 싱크 커넥터는 Kafka Connect에서 데이터를 가져온 다음 Elasticsearch API를 사용하여 데이터를 기록

소스 커넥터로부터 받은 데이터를 카프카 토픽에 작성하고 

카프카 토픽으로부터 데이터를 읽어 싱크 커넥터로 전달하는 것을 처리합니다. 

 

Transfer

원본 커넥터에서 Kafka에 기록되기 전에 데이터를 수정하고,

Kafka에서 읽은 데이터를 싱크에 기록하기 전에 수정합니다.

 

Transter 작업은 개별 메시지가 이동할 때 작동하므로

단일 메시지 변환(Single Message Transforms, SMT)으로 명명합니다.

 

Converter

컨버터는 Kafka Connect와 Kafka 간에 흐르는 데이터의 직렬화 및 역직렬화를 담당합니다.

키-값에 형식으로 컨버터를 지정합니다.

key.converter=org.apache.kafka.connect.storage.StringConverter
value.converter=org.apache.kafka.connect.storage.StringConverter
value.converter.schema.registry.url=http://localhost:8081

 

REST API를 사용하여 커넥터 인스턴스 추가하기

Kafka Connect의 REST API 호출로 Elasticsearch 싱크 커넥터 인스턴스를 생성하는 예

curl -X PUT -H  "Content-Type:application/json"\
http://localhost:8083/connectors/sink-elastic-01/config \
    -d '{
    "connector.class": "io.confluent.connect.elasticsearch.ElasticsearchSinkConnector",
    "topics"         : "orders",
    "connection.url" : "http://elasticsearch:9200",
    "type.name"      : "_doc",
    "key.ignore"     : "false",
    "schema.ignore"  : "true"
}'

 

ksqlDB로 커넥터 인스턴스 추가하기

ksqlDB를 사용하여 커넥터를 관리할 수도 있습니다.

이전 Elasticsearch 싱크 커넥터 인스턴스를 추가하기 위한 구문은 다음과 같습니다:

CREATE SINK CONNECTOR sink-elastic-01 WITH (
  'connector.class' = 'io.confluent.connect.elasticsearch.ElasticsearchSinkConnector',
  'topics'          = 'orders',
  'connection.url'  = 'http://elasticsearch:9200',
  'type.name'       = '_doc',
  'key.ignore'      = 'false',
  'schema.ignore'   = 'true'
);

 

 

 

 

참조 : https://developer.confluent.io/courses/kafka-connect/intro/

Kafka Connect Tutorial: How Connectors, Sinks & Sources Work