Stream your PostgreSQL changes into Kafka with Debezium

[@manojjoshi4321]

This is an extremely rich and high quality video. I stumbled upon this by chance when I was searching for some info about Debezium for some project we might be interested in. Interestingly enough, the files provided in the video description did not quite work for MAC and it took about 2 days to make the required adjustments including getting the correct images and port configuration, network configuration etc. but in the end things worked. My experience was that I learned a lot as I went through the process, not just about Debezium but a great deal of other things on what docker-compose structure, network dependency, port issues and all that. So it was worth. Kudos to Irtiza for taking all efforts to put together such as amazing tutorial....!!

[@irtizahafiz]

Thank you so much! Glad you folks got it working.

[@intruderstube]

Thanks, it is helpful. "Weird" thing with psql shell is just because of missing semicolon at the end.

[@irtizahafiz]

Ahh I see! That’s a good callout. I swear it used to work before without the semicolon :/ Glad you found it helpful regardless : )

[@bnssoftware3292]

Really like your teaching style. Thank you.

[@irtizahafiz]

Thanks for watching!

[@vjnt1star]

Very good presentation short and to the point. I got the information I needed. Thank you

[@irtizahafiz]

Glad you enjoyed it!

[@chrisschuck3316]

Awesome concise explanation. Appreciate your work!

[@irtizahafiz]

Thank you! I will start posting again soon, so please let me know what type of content interests you the most.

[@marwen1983-t1s]

thank you very much. i really appreciate your approach to discussing this topic

[@irtizahafiz]

You're very welcome!

[@RajkumarS-p1f]

did you the error docker run --tty --network postgres_debezium_cdc_default confluentinc/cp-kafkacat kafcakat -b kafka:9092 -C -s key=s -s value=avro -r schema-registry:8081 -t postgres.public.student" and get issue => docker: Error response from daemon: network postgres_debezium_cdc_default not found.

[@morph-87]

Thanks for the comprehensive explanation.

[@irtizahafiz]

You are welcome!

[@r-rtz]

Your 'Trevor' update is showing up in Kafka not as an update, but as a new record ("before": null)

[@wilfredomartel7781]

Just what i was searching for👏

[@irtizahafiz]

Glad : )

[@martingonzalez9298]

I have a question, when you did the update the event on the kafka topic had a '{"before": null, ' why is that null if the row already had information before ?

[@irtizahafiz]

Hi! That's a very good observation. I was actually running into that intermittently. Like 8/10 updates were working properly, but for 2/10 the before was NULL for some reason. Resetting the WAL of the database fixed it every time, but I am not sure why it happens in the first place.

[@iuriivdovin731]

@@irtizahafiz Hi! Before is null because you missed semicolon ';' after ALTER TABLE REPLICA IDENTITY FULL

[@marcinkiziukiewicz9695]

Thank you :)

[@SpiritOfIndiaaa]

Thanks a lot , simple and straight forward and clear

[@CuriouslyContent]

Any idea why the last update had a 'before' value of 'null'? Since the last thing was an update, shouldn't the 'before' have the values as they were before the update (id:2, name:mike)?

[@bulgakovwork2022]

same question

[@irtizahafiz]

Hi! Thank you for bringing this up. I think I answered this in another comment. I was running into this bug intermittently. I think it had to do with some misconfiguration of the postgres cluster.

[@manojjoshi4321]

I had the same issue but resolved it quickly. Make sure you use - alter table replica identity full.

[@geneschmidt9337]

THIS WAS GREAT! Thank you!

[@devamzanzmera2231]

It's very informative

[@irtizahafiz]

So nice of you

[@38vbharat]

Nice explanation ! Thanks

[@irtizahafiz]

Glad you found it helpful : )

[@sahilratnaparkhi5917]

can you make a video on streaming data from cassandra to kafka using debezium

[@irtizahafiz]

Noted. I will try to put something together.

[@sushantsinha4038]

Thanks ! Clear Explanation in this video . Do you have PostgreSQL Kafka Sink connector tutorial similar to this ?

[@irtizahafiz]

Hi! Unfortunately, I don't! I do plan on getting to it at some point though. Meanwhile, I am doing most of these data pipeline stuff using AWS. So if you are interested, please check out those videos.

[@vj_fantasy4298]

Hello actually I'm doing this currently and I'm stuck at 8:20 so can anyone please tell me what is --network and also what is postgres_debezium_default it's saying it doesn't exist

[@irtizahafiz]

Depends on the app you are building. I tried keeping things generic here. The ranking can be based on some score generated by a machine learning model, etc.

[@kurshy]

when i run the command to fire up dbz, i get "empty reply from server". kindly assist

[@SoumilShah]

Hello i need some help i was able to implement everything you showed in video i cannot consume messages from my local kafka-python can you please help ?

[@irtizahafiz]

Hi! It's difficult to help without more details.

[@mmfStudent]

Why after the sql update command, event has `before: null` in Kafka?

[@irtizahafiz]

Good question! I think I mention it in the video, I had to restart the Postgres instance to make it work.

[@UmarFarooq-u4g7h]

can i used this design for RAG system in production for real time update postgres to vector database?

[@irtizahafiz]

Yes you can.

[@alphonsetaylor2788]

Really helpful ! thanks

[@vijeandran]

Hi, Thanks for the video... Whenever I insert or update the data in postgres, those changes are not showing in the docker run command

[@irtizahafiz]

I would recommend restarting Postgres after you change its configuration.

[@vijeandran]

@@irtizahafiz Thanks you Sir... Now it is working as expected....

[@mukeshkumar-il3ct]

please help me I'm getting the error '% ERROR: Invalid token 'k' in key pack-format, see -s usage."

[@Malcolm777-i]

table.include.list: can this accept array or list separated by ,?

[@sanjaymadhavan1893]

How to deal with the concept of foreign key in debezium? Great video BTW! Life saviour

[@bnssoftware3292]

How come on the update it doesn't show the before data?

[@irtizahafiz]

That was happening intermittently. I believe there are some discussions about this in the comments section.

[@udaynj]

Nice video. Like your teaching style. Quick question - How stable is Debezium if you want to run this at scale in production?

[@irtizahafiz]

Many companies use it at scale.

[@AnkitKumar25]

I'm getting this error after running the final kafkacat command. I'm using windows to run this. I checked in both cmd and powershell. Error: file/topic list only allowed in producer(-P)/kafkaconsumer(-G) mode

[@AnkitKumar25]

docker run --tty --network bin_default confluentinc/cp-kafkacat kafkacat -b kafka:9092 -C -s key=s -s value=avro -r schema-registry:8081 -t postgres.public.student this command works for me now.

[@uma_mataji]

sorry when I ran the docker-compose command it just keeps on spinning forever ...it doesnot work.

[@irtizahafiz]

Hi! Sorry it's difficult for me to help without some more context.

[@pallavi1388]

Hi I am trying to add skipped.operations = "c" in debezium.json but its still showing create/update/delete?

[@irtizahafiz]

Hi! Unfortunately, I don't think I can be too helpful when it comes to debezium specifics.

[@priyantynurulfatimah6408]

Hi, so this is my JSON to create debezium connector but it returned error. Could you please help me troubleshoot? I am still learning and already look out for the solving but to no avail. Y^Y JSON: { "name": "source-transactions-connector", "config": { "connector.class": "io.debezium.connector.postgresql.PostgresConnector", "plugin.name": "pgoutput", "database.hostname": "postgres", "database.port": "5432", "database.user": "etl", "database.password": "demopass", "database.dbname": "dvdrental", "database.server.name": "postgres", "key.converter.schemas.enable": "false", "value.converter.schemas.enable": "false", "transforms": "unwrap", "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState", "key.converter": "org.apache.kafka.connect.json.JsonConverter", "value.converter": "org.apache.kafka.connect.json.JsonConverter", "table.include.list": "public.rental_streaming", "slot.name": "deb_rental_transaction_slot" } } ERROR: ERROR || Failed testing connection for jdbc:postgresql://postgres:5432/dvdrental with user 'etl': The connection attempt failed. [io.debezium.connector.postgresql.PostgresConnector]

[@pyramidofmerlinii4368]

If we have airflow in docker-compose, we don't need Debezium, right?

[@irtizahafiz]

I think you can do a lot with Airflow, but not sure about the specifics.

[@shubhamkapoor1123]

Hi, I am getting this error when running kafkacat command ERROR: Failed to format message in postgres.public.student [0] at offset 0: Avro/Schema-registry message deserialization: REST request failed (code -1): HTTP request failed: URL using bad/illegal format or missing URL : terminating Please tell the next step how to solve this. thanks!

[@irtizahafiz]

Not sure without more details. Can you provide with examples of your payload?

[@Bhavesh_Bhuvad]

docker: Error response from daemon: network postgres_debezium_cdc_master_default not found. i have getting this error on docker run -tty .....command.plz help

[@irtizahafiz]

Make sure you define the network in the docker-compose file.

[@pammyarw]

is there any way I can provide a topic name to which it should be published?

[@irtizahafiz]

Yes definitely!

[@inphosoftindonesia3694]

hi, i wonder, how about delete, does debezium also support delete?

[@irtizahafiz]

Yes. It supports delete too.

[@AnimeZone247]

Is it possible to listen to join queries?

[@irtizahafiz]

I believe CDC streams are generated only on data mutations, so no JOIN queries cannot be listened to by default.

[@fkaanoz]

I guess "ALTER TABLE ..." part didn't work, due to lack of semi column at the end. @4.20

[@mariemhajjem]

Thanks a lot!

[@irtizahafiz]

Glad it was helpful. I added the idea to my “future videos” list. The next few weeks I will be focusing on the System Design course, so unfortunately it will take some time :(

[@edisonngizwenayo5752]

Great explanations, just have few questions. The first one is how the configuration should be made in debezium.json file if the the source is an API other than Database. The last one, how to allow debezium to listen to change from multiple database tables. Thanks

[@irtizahafiz]

I can consider making future videos on these. Thank you!

[@RakantaRifky24]

If I want Yugabytedb postgresql for doing this migration, is it using same way like postgresql?

[@irtizahafiz]

It should be similar as long as there's a connector available.

[@RJ-wj7lc]

Nice video, do you have an example for Spring boot with Apache kafka and Debezium connector(MySQL, MS SQL Server)?

[@irtizahafiz]

Unfortunately, I do not. Maybe something in the future.

[@AkshitBansal-fk9gt]

nice tutorial. you missed ; in some commands, that's why the output was not visible.

[@irtizahafiz]

Sorry about that!

[@ansarhayat6276]

Hi Irtiza, any stuff related SQL server source and sink connectors

[@irtizahafiz]

I do have a few SQL videos coming up soon.

[@vaibhavpandey5615]

Great man ,could you please give me docker commond to start consumer on topic!

[@irtizahafiz]

Unfortunately I don't have the code for this anymore :(

[@PS-ff1tq]

Great introduction 👍,I have a question ,how to consume the data into a target database or consumer the data from topic using kafka python or any other api ?

[@irtizahafiz]

For consuming data into a target database, you should be able to do it with a Sink Connector of your respective database. Almost all Sink Connectors should be able to read from a Kafka topic. As for consuming Kafka using Python, I have an end to end example of that here in this video. I think you will find it helpful: kzbin.info/www/bejne/p5qapoVpftmUhbs

[@吳信宏-z7i]

Great introduction. Could you share your scripts in your video?

[@irtizahafiz]

Hi! I can share as much as my iterm command history allows me. If you need anything in particular, let me know. curl -i -X POST -H "Accept:application/json" -H "Content-Type:application/json" 127.0.0.1:8083/connectors/ --data "@debezium.json" docker run --tty --network postgres_debezium_cdc_default confluentinc/cp-kafkacat kafkacat -b kafka:9092 \ -C -s key=s -s value=avro -r schema-registry:8081 -t postgres.public.student

[@vilius518]

Thats weird, messages that end up in kafka topic are not in json, do you have clue why? Im using kafdrop for inspecting topics. Could it be that your kafkacat command also parses messages to json?

[@irtizahafiz]

Potentially. I know that the Kafkacat command has lots of "quality of life improvements" baked in.

[@parteeks9012]

Thanks, it helped me a lot but the weird thing is you updated the data in the database in the end, and still Kafka "before key" is null. Anyone has any thoughts on this?😁

[@irtizahafiz]

That has to do with some kind of Postgres configuration. I remember fixing it immediately after the video, but can't remember :(

[@rssaini138]

@@irtizahafiz It is working after below command, because in video it was not executed because you've missed semicolon. ALTER TABLE public.student REPLICA IDENTITY FULL;

[@nurulasyikin1895]

Is it possible to implement this without using docker environment? If it's possible can you demonstrate how to perform that?

[@irtizahafiz]

Hi! Yes it is possible! I might be able to demonstrate that some time in the future, but it’s currently not in my plans unfortunately.

[@rahadianpramono4050]

hi need your help, i try to run docker run --tty --network postgres_debezium_cdc_default but it showing ERROR: Failed to query metadata for topic postgres.public.student: Local: Broker transport failure, please help, thank you

[@irtizahafiz]

Are you sure the Kafka topic was created successfully? I would recommend listing the available topics in the broker first.

[@rahadianpramono4050]

@@irtizahafiz I'm not sure that kafka topic created successfully because I followed all the instructions in your video :D, how to listing the available topics in the broker? thank you

[@irtizahafiz]

You can run something like this in your CLI to list the Kafka topics: ./usr/bin/kafka-topics --zookeeper zookeeper:2181 --list

[@valentinadevanna9448]

@@irtizahafiz i have the same problem and the topic postgres.public.student exists

[@valentinadevanna9448]

it works!! It is sufficient runs the command two or threee times

[@prasadkintali6560]

Good one brother. Do you have a video to write to a source database from this Kafka source topic?

[@irtizahafiz]

I don’t have a video yet, but I do plan on making one soon. : )

[@neeraj2626]

@@irtizahafiz Did you happen to create one?

[@KshitijKohli-h8k]

Awesome tutorial! I wanted to understand 1 piece more. There are 2 patterns which I am now aware of of streaming CDC to Kafka. 1 via debezium connector that you talked about, other is via the Outbox pattern where the application service commits to an Outbox table in the same commit it writes to other application tables, post which tools like Ceres can stream the data to Kafka. What are the core differences in these 2 approaches and is there a recommendation of one over the other?

[@irtizahafiz]

I am actually not familiar with Outbox :(

[@JoseGuerra-qc5hq]

Typically the outbox pattern is to guarantee once delivery of the message, it's not to perform cdc (imagine a critical service that you must guarantee message delivery). Your application does not send messages to a kafka topic but instead send to a db table. The outbox service then is responsible to see which messages in your table (outbox) have been successfully published to kafka and retries for the ones who haven't.There is a chance that the outbox publishes a event more than once, so your consumers must be idempotent. From the consumers side you may have what would be called an inbox pattern, which blocks the same message to be consumer more than once, or they are okay with receiving the same message more than once. What you may have is CDC with Debezium but instead of publishing directly to kafka it publishes the events to a DB table via outbox.

[@HikarusVibrator]

Is there any way at all to do a major version DB upgrade without manually stopping writes, then killing the Debezium connector, etc. ? For a microservice architecture it really is quite costly to have such prolonged downtimes whenever a major upgrade is done. I don’t see a solution anywhere

[@irtizahafiz]

I don’t have any experience with it :( So unfortunately I won’t be of much help here sorry.

[@arnavdman]

I dont think so, the process of doing a major version upgrade of a database typically requires some level of downtime. But if you can setup a replica and divert the traffic there for the time being it might be a solution

[@darshankadiya2824]

can we use same using airflow

[@irtizahafiz]

Yup for sure! Products like AirFlow make data engineering super easy by abstracting away all these.

[@ranisari3110]

please make video postgresql -> debezium -> pubsub, thank you

[@irtizahafiz]

Hi! I do plan on doing that in the near future. After the system design videos that is.

[@dempile]

Even with the update , you get null ! ! ! !

[@irtizahafiz]

I know :( I was getting null every now and then. I believe it was because my Postgres docker container wasn't retaining the WAL setting every time I exited out of it.

[@brucewen9326]

In PostgreSQL, we often forgot to append ";" at the end of the SQL statement 🤪

[@irtizahafiz]

Thank you for the correction! I will start posting again soon, so please let me know what type of content interests you the most.

[@GordonShamway1984]

delete would be nice to see

[@bjugdbjk]

I am trying to build a project on top of this tutoral , got stuck below, could you help :) I ran the compose file and i can see from docker desktop all containers are up. From the go code i was trying to communicate with kafka to create a topic, throws me below error, could you help "panic: failed to dial: failed to open connection to kafka:9092: dial tcp: lookup kafka: no such host" Go code : conn, err := kafka.DialLeader(context.Background(), "tcp", "localhost:9092", "my-topic", 0) if err != nil { panic(err.Error()) } conn.SetReadDeadline(time.Now().Add(10 * time.Second))

[@irtizahafiz]

Hi! I am not really familiar with the project. Just make sure both the Go project and your kafka containers are running on the same network.

[@zaidahmed4254]

hi need your help, i try to run docker run --tty --network debezium_postgres_cdc_default(my app name start with debezium_postgres_cdc_default) but it showing ERROR: file/topic list only allowed in producer(-P)/kafkaconsumer(-G)

[@prasadkintali6560]

I have faced the same issue in windows command prompt, but it works in powershell.

[@irtizahafiz]

Are you trying it out in Windows or Mac or Linux?

[@robinthomas7350]

@@prasadkintali6560 i am having same issue in both command prompt and powershell.i am using windows machine. Can anyone help

[@srini1431]

Hi, I'm getting unable to find image 'confluentinc/cp-kafka:latest locally. How to resolve this please help

[@srini1431]

Error response from daemon: failed to create shim: OCI runtime create failed: container_linux.go:380: starting container process caused: exec: "-b": executable file not found in $PATH: unknown

[@irtizahafiz]

Hi! I can look at this later on in the week when I have some time.

[@atharvadeokar2814]

Hi Irtiaz I am facing issue - ERROR: Failed to query metadata for topic postgres.public.student: Local: Broker transport failure ./usr/bin/kafka-topics --zookeeper zookeeper:2181 --list This gives -> -bash: ./usr/bin/kafka-topics: No such file or directory I will really appreciate you if you could help me this issue Thankyou!

[@irtizahafiz]

Have you created the Kafka topics successfully? You have to register the topic first, before you can consume from it.

[@venkateshajjs5014]

docker: Error response from daemon: network postgres_debezium_cdc_default not found. having this error while run this command : docker run --tty --network postgres_debezium_cdc_default confluentinc/cp-kafkacat -b kafka:9092 -C -s key=s value=avro -r schema-registry:8001 -t postgres.public.student ...

[@venkateshajjs5014]

please help me .. --network postgres_debezium_cdc_default denotes what?

[@venkateshajjs5014]

please i need help...

[@iuriivdovin731]

That's because the network has another name. Print in the command line "docker network ls" and find the network in the list. In my case it was "debezium-default".

[@venkateshajjs5014]

@@iuriivdovin731 thanks bro

[@irtizahafiz]

Thanks for helping out! : )

[@christopherma5445]

I got the following error :% ERROR: Topic postgres.public.student error: Broker: Leader not available after running docker run --tty --network postgres_debezium_cdc_default confluentinc/cp-kafkacat kafkacat -b kafka:9092 -C -s key=s -s value=avro -r schema-registry:8081 -t postgres.public.student

[@irtizahafiz]

Maybe try resetting Docker if it's running already?