[Kafka] Kafka Connect 날짜 변환 문제 해결 - Avro / Epoch

문제는 바야흐로 7개월 전...
kafka와 elastic을 활용한 모니터링 대시보드를 기획하여 구축하였다.
관련 기술들에 아무런 배경지식이 없이 시작한 나는 뚱땅뚱땅 구축 성공처럼 보였으나 문제가 있었다.
Kibana를 통해 시각화를 하려고 하니 날짜 데이터가 이상하게 조회가 되는 것이 아닌가!

문제의 원인을 파악하고,, 다시 개념부터 살펴보고,, 삽질의 삽질을 거쳐 드디어 해결하였다.
중간에 다른 업무로 놓은 시기도 있었지만 장장 4개월만에 해결한거라 감회가 새롭다^^

물론 아주 기본적인, 다른 사람들은 처음부터 잘 알고 있는 부분일 수도 있지만
나와 같이 헤매다 이쪽으로 오게 되신 분 들도 계실테니
삽질로그를 남겨보겠다.

---

 

문제 상황

SoureDB의 "2024-01-08" date값이 '19730'과 같은 Epoch day값으로 마음대로 변환이 되어 메시징 되는 경우

해결 방법

data = {
    "name": "<my-source-connector>",
    "config": {
        "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
        "connection.url": "jdbc:postgresql://<postgres-host>:5432/<database>",
        "connection.user":"<username>",
        "connection.password":"<password>",
        "mode":"incrementing",
        "incrementing.column.name" : "<seq>",
        "table.whitelist" : "<schema.table_name>",
        "topic.prefix" : "<schema_>",
        "tasks.max" : "1",
        "transforms": "convertTimestamp",
        "transforms.convertTimestamp.type": "org.apache.kafka.connect.transforms.TimestampConverter$Value",
        "transforms.convertTimestamp.format": "yyyy-MM-dd",
        "transforms.convertTimestamp.target.type": "string",
        "transforms.convertTimestamp.field": "<created_at>"
    }
}

< >안의 값은 실제 환경에 맞게 수정!

Kafka Connect에서는 데이터 변환을 수행할 수 있는 SMT(Single Message Trasform)라는 기능을 제공한다.
이를 통해 시간 형식을 지정하여 변환할 수 있다. 코드의 밑 부분에 있는 transform 부분을 원하는 형식으로 명시하여 작성하면 된다!

나의 경우에는 JSON형식의 데이터를 AVRO로 변환을 하겠다고 지정을 해준 것이다.
위 옵션을 추가해주면 기존에 Epoch days의 날짜로 전달이되어서 Elasticsearch에서 date로 인식을 하지 못했는데 "yyyy-MM-dd"로 잘 전달이 되는 결과를 볼 수 있다!!

 

추가 개념

이를 해결하는 과정에서 알게 된 개념들을 정리해보겠다.

19730이 대체 무슨 의미일까? 에서부터 시작된다.

Epoch

Epoch는 시간을 측정하는 일반적인 방법 중 하나로, Unix Epoch라고 하는 1970년 1월1일 00:00:00:UTC를 기준으로 한 시간의 경과를 나타낸다.

19730이란 1970년 1월1일 부터 19730일 지났다는 값인 것이다.

AVRO

Avro는 데이터 직렬화 형식 중 하나로, 프레임워크 간 데이터 교환을 위해 사용된다.
Apache Avro 프로젝트에서 관리되고 있다.

Elasticsearch에 데이터를 전송할 때 Avro 포맷을 사용하는 것이 일반적이고 
"Date"를 "Int"타입으로 표현되며, Unix Epoch로부터의 일수를 나타낸다.

ISO

ISO는 날짜와 시간을 표현하기 위한 국제 표준 형식으로 우리가 흔히 알고 있는 형식이다.
날짜 형식: YYYY-MM-DD
날짜 및 시간 형식: YYYY-MM-DDTHH:MM:SS

Kakfa Connect를 사용하여 데이터를 다른 시스템으로 전송할 때, 보통 데이터 형식을 표준화하고 효율적인 직렬화 및 역직렬화를 위해 Avro와 같은 변환기를 사용하게 된다.
하지만 나는 elasticsearch에서 avro->iso 형식으로 매핑이 잘 되지 않아 string형식으로 format하여 전송하는 방식을 사용한 것이다!

 

삽질로그

• Avro변환을 사용하지 않고 JSON 형식으로 데이터를 유지하는 설정
  
  

"value.converter": "org.apache.kafka.connect.json.JsonConverter",
"key.converter": "org.apache.kafka.connect.json.JsonConverter",
"key.converter.schemas.enable": "false",
"value.converter.schemas.enable": "false"

• Long타입으로 데이터 보내고 Elasticsearch에서 format

# source connect
data = {
	...
	"config" = {
		...
        "schema.mapping": "created_at:long"
	}
}

# elastic search
PUT 인덱스이름
{
	...
    "mappings" : {
     "properties" : {
     	...
        "created_at": {
        "type": "date",
        "format": "strict_date_optional_time||epoch_second"
      }
    }
 }

• server.properties 설정 이리저리 바꿔보기...

 

7.2.3 날짜 - date - Elastic 가이드북 (kimjmin.net)

7.2.3 날짜 - date | Elastic 가이드북

형식 | Elasticsearch 가이드 [8.12] | 탄력 있는

format | Elasticsearch Guide [8.12] | Elastic

Date format error with "epoch_millis||epoch_second" - Elastic Stack / Elasticsearch - Discuss the Elastic Stack

Date format error with "epoch_millis||epoch_second"