🎄 12일 간의 SMT 🎄 - 8일 차: TimestampConverter

TimestampConverter 단일 메시지 변환을 사용하면 Kafka 메시지의 타임스탬프 필드로 작업할 수 있습니다. 문자열을 기본Timestamp 유형(또는 Date 또는 Time )과 Unix epoch로 변환할 수 있으며 그 반대도 마찬가지입니다.

이것은 Kafka에 수집된 데이터가 타임스탬프로 올바르게 저장되었는지 확인하는 데 매우 유용하며(있는 경우) 타임스탬프를 선택한 문자열 형식으로 싱크 커넥터에 쓸 수도 있습니다.
TimestampConverter는 세 개의 인수를 취합니다. 타임스탬프가 있는 필드의 이름, 변환하려는 데이터 유형, 구문 분석할 타임스탬프 형식(문자열 및 target.type을 읽는 경우: unix/timestamp/date/time ) 또는 쓰기(문자열을 작성 중이고 target.type가 string인 경우). 타임스탬프 형식은 SimpleDateFormat 을 기반으로 합니다.

"transforms" : "convertTS",
"transforms.convertTS.type" : "org.apache.kafka.connect.transforms.TimestampConverter$Value",
"transforms.convertTS.field" : "txn_date",
"transforms.convertTS.format" : "EEE MMM dd HH:mm:ss zzz yyyy",
"transforms.convertTS.target.type": "Timestamp"

👾 Demo code

문자열 기반 타임스탬프를 타임스탬프로 변경

페이로드에서 txn_date 필드는 타임스탬프처럼 보입니다.

$ docker exec kafkacat kafkacat -b broker:29092 -r http://schema-registry:8081 -s key=s -s value=avro -t day8-transactions -C -c5 -o-5 -u -q -J | \
  jq '.payload.Gen0.txn_date.string'

"Thu Dec 10 17:06:59 GMT 2020"
"Sat Dec 05 16:39:40 GMT 2020"
"Sat Dec 05 21:43:46 GMT 2020"
"Sun Dec 13 20:30:21 GMT 2020"
"Wed Dec 09 06:18:31 GMT 2020"

그러나 값에 대한 실제 스키마를 살펴보면(Avro로 직렬화되므로 Protobuf 또는 JSON 스키마에도 동일하게 적용됨) 타임스탬프처럼 보일 수 있고 타임스탬프처럼 돌팔이처럼 보일 수 있지만 실제로는 타임스탬프 자체가 아니라 문자열:

$ curl -s "http://localhost:8081/subjects/day8-transactions-value/versions/latest" | jq '.schema|fromjson[]'

"null"
{
  "type": "record",
  "name": "Gen0",
  "namespace": "io.mdrogalis",
  "fields": [
[…]
    {
      "name": "txn_date",
      "type": [
        "null",
        "string"
      ]
[…]

즉, Kafka Connect 싱크와 같은 소비자가 데이터를 사용할 때 대상 개체 유형이 일반적으로 동일하게 상속하는 결과와 함께 여전히 문자열로 처리됩니다. 다음은 JDBC sink connector ( 🎥 Kafka Connect in Action : JDBC Sink (👾 demo code ) 및 🎥 ksqlDB & Kafka Connect JDBC Sink in action (👾 demo code ))의 예입니다.

curl -i -X PUT -H "Accept:application/json" \
    -H "Content-Type:application/json" http://localhost:8083/connectors/sink-jdbc-mysql-day8-00/config \
    -d '{
          "connector.class" : "io.confluent.connect.jdbc.JdbcSinkConnector",
          "connection.url" : "jdbc:mysql://mysql:3306/demo",
          "connection.user" : "mysqluser",
          "connection.password": "mysqlpw",
          "topics" : "day8-transactions",
          "tasks.max" : "4",
          "auto.create" : "true",
          "auto.evolve" : "true"
        }'

mysql> describe `day8-transactions`;
+------------------+------+------+-----+---------+-------+
| Field            | Type | Null | Key | Default | Extra |
+------------------+------+------+-----+---------+-------+
| customer_remarks | text | YES  |     | NULL    |       |
| item             | text | YES  |     | NULL    |       |
| cost             | text | YES  |     | NULL    |       |
| card_type        | text | YES  |     | NULL    |       |
| txn_date         | text | YES  |     | NULL    |       |
+------------------+------+------+-----+---------+-------+
5 rows in set (0.00 sec)

txn_date는 text 필드로 타임스탬프로 사용하려는 사람에게는 아무 소용이 없습니다.

여기에서 TimestampConverter 이 나옵니다. 이 예에서는 제공된 날짜 및 시간 형식 문자열을 사용하여 Kafka Connect를 통과할 때 문자열을 타임스탬프로 캐스팅합니다. 또한 이를 사용하여 에포크 타임스탬프 값 간에 변환하고 문자열, 에포크, 날짜 또는 시간(실제 타임스탬프도 포함)을 대상으로 지정할 수 있습니다.

curl -i -X PUT -H "Accept:application/json" \
    -H "Content-Type:application/json" http://localhost:8083/connectors/sink-jdbc-mysql-day8-01/config \
    -d '{
          "connector.class" : "io.confluent.connect.jdbc.JdbcSinkConnector",
          "connection.url" : "jdbc:mysql://mysql:3306/demo",
          "connection.user" : "mysqluser",
          "connection.password" : "mysqlpw",
          "topics" : "day8-transactions",
          "tasks.max" : "4",
          "auto.create" : "true",
          "auto.evolve" : "true",
          "transforms" : "convertTS,changeTopic",
          "transforms.convertTS.type" : "org.apache.kafka.connect.transforms.TimestampConverter$Value",
          "transforms.convertTS.field" : "txn_date",
          "transforms.convertTS.format" : "EEE MMM dd HH:mm:ss zzz yyyy",
          "transforms.convertTS.target.type": "Timestamp",
          "transforms.changeTopic.type" : "org.apache.kafka.connect.transforms.RegexRouter",
          "transforms.changeTopic.regex" : "(.*)",
          "transforms.changeTopic.replacement": "$1_withTS"
        }'

다음은 MySQL의 결과 테이블입니다.

mysql> describe `day8-transactions_withTS`;
+------------------+-------------+------+-----+---------+-------+
| Field            | Type        | Null | Key | Default | Extra |
+------------------+-------------+------+-----+---------+-------+
| customer_remarks | text        | YES  |     | NULL    |       |
| item             | text        | YES  |     | NULL    |       |
| cost             | text        | YES  |     | NULL    |       |
| card_type        | text        | YES  |     | NULL    |       |
| txn_date         | datetime(3) | YES  |     | NULL    |       |
+------------------+-------------+------+-----+---------+-------+
5 rows in set (0.00 sec)

위에서 언급했듯이 target.type 를 변경하여 타임스탬프의 날짜 또는 시간 구성 요소만 추출할 수도 있습니다.

날짜만

curl -i -X PUT -H "Accept:application/json" \
    -H "Content-Type:application/json" http://localhost:8083/connectors/sink-jdbc-mysql-day8-02/config \
    -d '{
          "connector.class" : "io.confluent.connect.jdbc.JdbcSinkConnector",
          "connection.url" : "jdbc:mysql://mysql:3306/demo",
          "connection.user" : "mysqluser",
          "connection.password" : "mysqlpw",
          "topics" : "day8-transactions",
          "tasks.max" : "4",
          "auto.create" : "true",
          "auto.evolve" : "true",
          "transforms" : "convertTS,changeTopic",
          "transforms.convertTS.type" : "org.apache.kafka.connect.transforms.TimestampConverter$Value",
          "transforms.convertTS.field" : "txn_date",
          "transforms.convertTS.format" : "EEE MMM dd HH:mm:ss zzz yyyy",
          "transforms.convertTS.target.type": "Date",
          "transforms.changeTopic.type" : "org.apache.kafka.connect.transforms.RegexRouter",
          "transforms.changeTopic.regex" : "(.*)",
          "transforms.changeTopic.replacement": "$1_withDate"
        }'

MySQL의 결과 테이블:

mysql> describe `day8-transactions_withDate`;
+------------------+------+------+-----+---------+-------+
| Field            | Type | Null | Key | Default | Extra |
+------------------+------+------+-----+---------+-------+
| customer_remarks | text | YES  |     | NULL    |       |
| item             | text | YES  |     | NULL    |       |
| cost             | text | YES  |     | NULL    |       |
| card_type        | text | YES  |     | NULL    |       |
| txn_date         | date | YES  |     | NULL    |       |
+------------------+------+------+-----+---------+-------+
5 rows in set (0.01 sec)

mysql> select txn_date from `day8-transactions_withDate` LIMIT 5;
+------------+
| txn_date   |
+------------+
| 2020-01-04 |
| 2020-01-04 |
| 2019-12-29 |
| 2020-01-01 |
| 2019-12-29 |
+------------+
5 rows in set (0.00 sec)

시간만

curl -i -X PUT -H "Accept:application/json" \
    -H "Content-Type:application/json" http://localhost:8083/connectors/sink-jdbc-mysql-day8-03/config \
    -d '{
          "connector.class" : "io.confluent.connect.jdbc.JdbcSinkConnector",
          "connection.url" : "jdbc:mysql://mysql:3306/demo",
          "connection.user" : "mysqluser",
          "connection.password" : "mysqlpw",
          "topics" : "day8-transactions",
          "tasks.max" : "4",
          "auto.create" : "true",
          "auto.evolve" : "true",
          "transforms" : "convertTS,changeTopic",
          "transforms.convertTS.type" : "org.apache.kafka.connect.transforms.TimestampConverter$Value",
          "transforms.convertTS.field" : "txn_date",
          "transforms.convertTS.format" : "EEE MMM dd HH:mm:ss zzz yyyy",
          "transforms.convertTS.target.type": "Time",
          "transforms.changeTopic.type" : "org.apache.kafka.connect.transforms.RegexRouter",
          "transforms.changeTopic.regex" : "(.*)",
          "transforms.changeTopic.replacement": "$1_withTime"
        }'

MySQL의 결과 테이블:

mysql> describe `day8-transactions_withTime`;
+------------------+---------+------+-----+---------+-------+
| Field            | Type    | Null | Key | Default | Extra |
+------------------+---------+------+-----+---------+-------+
| customer_remarks | text    | YES  |     | NULL    |       |
| item             | text    | YES  |     | NULL    |       |
| cost             | text    | YES  |     | NULL    |       |
| card_type        | text    | YES  |     | NULL    |       |
| txn_date         | time(3) | YES  |     | NULL    |       |
+------------------+---------+------+-----+---------+-------+
5 rows in set (0.00 sec)

mysql> select txn_date from `day8-transactions_withTime` LIMIT 5;
+--------------+
| txn_date     |
+--------------+
| 14:05:19.000 |
| 14:09:11.000 |
| 19:18:25.000 |
| 03:22:06.000 |
| 09:57:44.000 |
+--------------+
5 rows in set (0.00 sec)

유닉스 시대

유닉스 시대를 작성할 수도 있습니다.

curl -i -X PUT -H "Accept:application/json" \
    -H "Content-Type:application/json" http://localhost:8083/connectors/sink-jdbc-mysql-day8-04/config \
    -d '{
          "connector.class" : "io.confluent.connect.jdbc.JdbcSinkConnector",
          "connection.url" : "jdbc:mysql://mysql:3306/demo",
          "connection.user" : "mysqluser",
          "connection.password" : "mysqlpw",
          "topics" : "day8-transactions",
          "tasks.max" : "4",
          "auto.create" : "true",
          "auto.evolve" : "true",
          "transforms" : "convertTS,changeTopic",
          "transforms.convertTS.type" : "org.apache.kafka.connect.transforms.TimestampConverter$Value",
          "transforms.convertTS.field" : "txn_date",
          "transforms.convertTS.format" : "EEE MMM dd HH:mm:ss zzz yyyy",
          "transforms.convertTS.target.type": "unix",
          "transforms.changeTopic.type" : "org.apache.kafka.connect.transforms.RegexRouter",
          "transforms.changeTopic.regex" : "(.*)",
          "transforms.changeTopic.replacement": "$1_withUnixEpoch"
        }'

MySQL의 결과 테이블:

mysql> describe `day8-transactions_withUnixEpoch`;
+------------------+--------+------+-----+---------+-------+
| Field            | Type   | Null | Key | Default | Extra |
+------------------+--------+------+-----+---------+-------+
| customer_remarks | text   | YES  |     | NULL    |       |
| item             | text   | YES  |     | NULL    |       |
| cost             | text   | YES  |     | NULL    |       |
| card_type        | text   | YES  |     | NULL    |       |
| txn_date         | bigint | YES  |     | NULL    |       |
+------------------+--------+------+-----+---------+-------+
5 rows in set (0.00 sec)

mysql> select txn_date from `day8-transactions_withUnixEpoch` LIMIT 5;
+---------------+
| txn_date      |
+---------------+
| 1577973919000 |
| 1577714951000 |
| 1577819905000 |
| 1577762526000 |
| 1577786264000 |
+---------------+
5 rows in set (0.00 sec)

유닉스 에포크(bigint)의 타임스탬프가 소스로 있는 경우 TimestampConverter를 사용하여 타임스탬프/날짜/시간 및 문자열로도 쓸 수 있습니다. 후자를 수행하면 format 구성이 다음에 적용됩니다. 문자열이 기록될 형식입니다.

중첩 필드의 타임스탬프 액세스

불행히도 TimestampConverter는 루트 수준 요소에서만 작동합니다. 다른 필드에 중첩된 타임스탬프 필드에는 사용할 수 없습니다. 먼저 Flatten 을 사용하거나 고유한 변환을 작성해야 합니다.

그것을 밖으로 시도!

Docker Compose를 포함하여 이를 시도하기 위한 전체 코드를 찾을 수 있으므로 로컬 시스템에서 실행할 수 있습니다👾 here.