No video :(

🎄 Twelve Days of SMT 🎄 - Day 1: InsertField (timestamp)

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

Просмотров 10 тыс.

50% 1

Kafka Connect is the primary method for doing integration with Apache Kafka and other systems. It includes Single Message Transform (SMT) which can be used to modify messages as they pass through the pipeline in or out of Kafka.
This video explains using the InsertField SMT to insert the timestamp of each message into its payload when sending it to a sink such as AWS S3 or MySQL.
---
✍️Blog: rmoff.net/2020/12/08/twelve-d...
👾 Demo code: github.com/confluentinc/demo-...
Learn more about Kafka Connect here:
🏃‍♂️ Quick: • Kafka Connect in 60 se...
🚶 More detail: rmoff.dev/kafka-connect-zero-...
---
⏱ Time codes:
00:00:00 Brief introduction to Kafka Connect and Single Message Transforms
00:01:11 GitHub repository for the demo
00:01:47 Creating the source data generator connector (Voluble)
00:02:22 Examining the topics created by the source connector
00:03:08 Kafka Message timestamp metadata
00:04:02 Streaming data from Kafka to MySQL with the JDBC Sink connector
00:05:19 Adding in Kafka message timestamp to a sink connector
00:05:58 Using the InsertField Single Message Transform (SMT)
00:09:13 Streaming data from Kafka to S3
00:09:50 Using Transforms with the S3 sink to add the timestamp in string format
---
☁️ 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

Ссылка:

Скачать:

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

Добавить в:

Мой плейлист

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

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

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

thank you this really helps and is indeed a very powerful feature to be aware of

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

Worked like a charm !! Thanks @Robin

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

I'm debating whether spinning up a new microservice to handle consume and persist to a datastore, or do a kafka jdbc sink. As I don't have too much experience using a kafka sink, this gave me insight more insight to make proper decision. Thanks Robin.

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

Glad it was useful!

@anuragtelang Год назад

Hi Robin, Thanks for the tutorials on Kafka connect. These helped immensely during my POC. I am currently dealing with a problem where Kafka JDBC source connector reads a nested json from DB and it needs to map it to the Avro applied on kafka topic. I have posted a question on stackoverflow (Search with "Kafka JDBC source connector for Oracle DB - How to map nested json to avro schema"). Would be grateful if you could answer it. Thanks

@chintanthakker4129 4 месяца назад

I tried in MongoSinkConnector, I am seeing a failure with this message DataException: Only Map objects supported in absence of schema for [field insertion], fund java.lang.String \t in requireMap. Correct me, do I need to use value.convertor with value as JsonConvertor?

@Abhijit_Patra 4 месяца назад

I am trying to do this but I get a null value.

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

Is there a way I can insert a time stamp field when I sink data into Elastic Search? I’ve tried to sink data into Elastic Search but when it comes to querying in the index, I don’t have the option to query by that time stamp field.

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

I've made a video just about this :) Check out the tutorial notes here: github.com/confluentinc/demo-scene/tree/master/kafka-to-elasticsearch#timestamps and the video itself here: ru-vid.com/video/%D0%B2%D0%B8%D0%B4%D0%B5%D0%BE-Cq-2eGxOCc8.html For more questions, head to forum.confluent.io/

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

Hi Robin, wondering if I can combine this SMT with the convert SMT. Use-case. A record has a Timestamp string field. Which I can transform into an INT64 with a SMT. However I just wish to add the Timestamp INT64 convertion to the record (ie field: 'my_ts'), without having to convert the original field. Is it possible to achieve this using InsertField SMT or any other SMT ? Thanks and thanks for the great content you publish

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

hi Alessandro - can you post this at forum.confluent.io/ and I or someone else will try and answer for you there

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

@@rmoff Thanks for the fast feedback. Much appreciated. Posted my ask in the forum. See you there

@SatyendraSingh-ov1de 2 года назад

Hi Robin, I have use case where i want to pass schema name i.e table name from schema in to my payload or headers. Is there any way to do this, without enabling schema

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

Hi, I'm not sure if that is possible. You could try asking on forum.confluent.io/.

@SatyendraSingh-ov1de 2 года назад

@@rmoff Is there a way to extract a field from schema?

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

I have a requirement to feed my data in topic to elasticsearch. My messages have a timestamp field (both in long-milliseconds and date format (yyyyMMdd). I want to use one of the field to set @timestamp field in elasticsearch for indexing. How to do this?

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

Hi Sarath, a good place to ask this question is at forum.confluent.io/ :)

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

I used this SMT and the new field was created but it was null. MySQL Source, debezium connector and Bigqeuery Sink. What am I doing wrong?

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

This isn't the best place to debug - head over to cnfl.io/slack and post full details of your connector config, and also whether it's at the ingest or egress that you're seeing the null field.