🎄Twelve Days of SMT 🎄 - Day 8: TimestampConverter

Подписаться 4,2 тыс.

Просмотров 1,6 тыс.

50% 1

The TimestampConverter Single Message Transform lets you work with timestamp fields in Kafka messages. You can convert a string into a native Timestamp type (or Date or Time), as well as Unix epoch - and the same in reverse too.
This is really useful to make sure that data ingested into Kafka is correctly stored as a Timestamp (if it is one), and also enables you to write a Timestamp out to a sink connector in a string format that you choose.
The TimestampConverter takes three arguments; the name of the field holding the timestamp, the data type to which you want to convert it, and the format of the timestamp to parse (if reading a string and target.type is unix/timestamp/date/time) or write (if you’re writing a string and target.type is string). The timestamp format is based on 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"
---
⏱️Time codes:
00:00:00 Brief introduction
00:01:16 Kafkacat - the swiss-army knife of Kafka commandline tools
00:02:45 Inspecting the schema for a topic in the Schema Registry
00:04:56 Setting up the basic sink connector (no transform)
00:05:38 Using the TimestampConverter to write a string to a timestamp in a database table
00:08:56 Write the date portion of a string timestamp to a Date type in target sink connector
00:10:54 Write the date portion of a string timestamp to a Date type in target sink connector
00:11:43 Converting a string timestamp into unix timestamp (epoch)
00:14:10 Using the TimestampConverter with a Source connector
00:17:00 A brief tangent about useful bash keyboard shortcuts
00:18:19 Converting at ingest vs egress
00:18:50 Recap
---
👾 Demo code and details: github.com/confluentinc/demo-...
🗒️ SMT reference: docs.confluent.io/platform/cu...
Learn more about Kafka Connect here:
🏃‍♂️ Quick: • Kafka Connect in 60 se...
🚶 More detail: rmoff.dev/kafka-connect-zero-...
---
☁️ Confluent Cloud: confluent.cloud/signup?...
💾Download Confluent Platform: www.confluent.io/download/?ut...
📺 Kafka Connect connector deep-dives: • Kafka Connect
✍️Kafka Connect documentation: docs.confluent.io/current/con...
🧩Confluent Hub: www.confluent.io/hub/?...

Наука

Опубликовано:

3 авг 2024

Ссылка:

Скачать:

Готовим ссылку...

Добавить в:

Мой плейлист

Посмотреть позже

Комментарии : 14

@samiketabi53 2 года назад

Hi, thanks for this great video I use debezium for lead changes to Kafka topic and want to convert epoch integer base to TimeStamp after adding jar files to use in connector source config it does not work correctly

@rmoff 2 года назад

Please ask this question at forum.confluent.io/. Thanks!

@kevinjang8209 3 года назад

I am using the Elastic Search Sink. I just got my message time stamp set into a string since the time stamp wasn’t originally in there. Am I able to now convert that string into a time stamp in that same connector?

@rmoff 3 года назад

I'm not sure I quite follow your question. Head over to forum.confluent.io/ and I can help out there.

@lokeishdesaichetty1198 2 года назад

I have used 'w' for week no in kafka-connect configuration, im getting week no as 6 but actually the week no is 5 (Feb 02 2022), what is the correct way to get the week number from timestamp(my usecase is to add week no to the elasticsearch index and push data.)

@rmoff 2 года назад

I don't know the answer - please ask at forum.confluent.io/ and someone should be able to help. Thanks.

@anggaraajiprasetya6356 2 года назад

Hello Sir, I'm usin debezium for source sql, the schema type in avro when I sink it into sql server it worked transform into timestamps but when I sink it into mongodb it doesn't work, still get value base int64 in my timestamp field, how to transform that in mongo ?

@rmoff 2 года назад

hi, please head to forum.confluent.io/ and ask there :) thanks.

@abhrasarkar8040 3 года назад

Hello Sir, One quick doubt. I have a source table with a column that has integer values. Now the thing is I want to check that column's value and depending upon that I want to change the topic value (Assume I am using debezium source connector). Example Source Table A -> Column A 1 0 3 4 Sink Table B -> Column B 0 1 4 3 If source columnA's value is 1 then I want to insert 0 into sink ColumnB's value and so on. Thanks in advance!

@rmoff 3 года назад

Sounds like a good usecase for Kafka Streams or ksqlDB :)

@abhrasarkar8040 3 года назад

@@rmoff Thank You Sir

@abhrasarkar8040 3 года назад

@@rmoff Another small doubt sir. I have multiple datetime columns in source table. I want one column1 to be converted in unix and column2 into date in one source connector (Debezium). What will be the code structure sir? Thank you so much!