02.SingleMessageTransform(SMT),Converter

Connect Worker

Connector를 배포/구동 시키는 프로세스

Connector 배포 전의 모습

Connector 배포

Connecotr Class 지정

Connector 배포를 위한 Configuration의 일부

"config": {
[...]
"connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
"connection.url": "jdbc:postgresql://postgresql_ip:5432/",
"topics": ”sample.orders",
}

Connect Worker에 Connector Instance 및 Task 생성

• Connector Task가 Source 시스템에서 데이터를 가져와서 Connect Record로 변환

• 대상 시스템을 연결하기 위한 Connector가 없다면, Connect Framework를 기반으로 Custom Connector 개발 가능

Converter 설정

Key와 Value에 각각 Converter 설정

key.converter=io.confluent.connect.avro.AvroConverter
key.converter.schema.registry.url=http://localhost:8081
value.converter=io.confluent.connect.avro.AvroConverter
value.converter.schema.registry.url=http://localhost:8081

• 설정한 Converter가 Connect Record를 Byte Array로 변환한 후, Kafka로 전달

• Converter도 필요시 custom개발 가능

Single Message Transform(SMT)

Task와 Converter 사이에서 데이터 변환이 필요한 경우 사용

Transform	Description
Cast	필드 또는 전체 Key 또는 Value를 특정 유형으로 캐스트(타입 변환) (예: 정수 필드를 더 작은 너비로 강제 적용)
Drop	레코드에서 Key 또는 Value를 삭제하고 null로 설정
InsertField	레코드 메타데이터 또는 구성된 Static Value의 속성을 사용하여 필드를 삽입
MaskField	필드 유형에 대해 유효한 null 값으로 지정된 필드를 마스킹
ReplaceField	필드를 필터링하거나 이름을 변경

• 약 20개 정도의 Pre-defined SMT 제공

SMT 설정

"config": {
[...]
"transforms": "addTimeToTopic,labelBar",
"transforms.addTimeToTopic.type": "org.apache.kafka.connect.transforms.TimestampRouter",
"transforms.addTimeToTopic.topic.format": "${topic}-${timestamp}",
"transforms.addTimeToTopic.timestamp.format": "YYYYMMDD",
"transforms.labelBar.type": "org.apache.kafka.connect.transforms.ReplaceField$Value",
"transforms.labelBar.renames": "delivery_addr:shipping_addr",
}

• 여러 개의 SMT을 연결(Chaining)하여 사용 가능

단건 메시지별 데이터 변환 기능

• Connect Record를 설정된 SMT의 순서에 따라 변환

• SMT도 필요시 custom 개발 가능

• Transform을 자주하게 되면 CPU 사용률을 고려해야한다.

Sink Connector의 Data Flow

• Sink Connector의 Data Flow는 Source Connector의 역방향