<\/div>\n

Merhaba,<\/p>\n

Bu i\u00e7eri\u011fimizde birden \u00e7ok veri kayna\u011f\u0131nda ger\u00e7ek zamanl\u0131 olu\u015fan de\u011fi\u015fikliklerin yakalanmas\u0131n\u0131 ve stream edilmesini sa\u011flayan, open source ve distributed bir ara\u00e7 olan Debezium CDC(Change Data Capture)’yi inceliyor olaca\u011f\u0131z.<\/p>\n

CDC(Change Data Capture) Nedir?<\/h4>\n
Veritaban\u0131nda insert, update ve delete i\u015flemleri neticesinde nihai olarak \u00f6ncesinden farkl\u0131 olan, bir ba\u015fka deyi\u015fle de\u011fi\u015fiklik g\u00f6steren verileri yakalamam\u0131z\u0131, izlememizi ve i\u015flem yapabilmemizi sa\u011flayan bir tasar\u0131m modelidir. Bu model sayesinde verilerin ilk ve son hallerini elde edebilir ve t\u00fcrl\u00fc operasyonlar ger\u00e7ekle\u015ftirebiliriz. Tabi ki de CDC’yi kullanabilmek i\u00e7in ilgili veritaban\u0131 taraf\u0131ndan destekleniyor olmas\u0131 gerekmektedir.<\/p>\n
La hoca! Veritaban\u0131ndaki verisel de\u011fi\u015fiklikleri izleyebilmek i\u00e7in trigger kullanamaz m\u0131y\u0131z?<\/em><\/strong>
\nTabi ki de trigger ile bu i\u015flemleri ger\u00e7ekle\u015ftirebilirsiniz. Lakin CDC ile ihtiyaca d\u00f6n\u00fck operasyonlar\u0131 daha h\u0131zl\u0131 bir \u015fekilde i\u015fleyebilirsiniz. \u00c7\u00fcnk\u00fc CDC verilere dair t\u00fcm bilgileri veritaban\u0131 loglar\u0131ndan toplayarak hizmet sa\u011flamaktad\u0131r. Bu sebepten dolay\u0131 trigger’lara nazaran daha performansl\u0131 ve az maliyetli bir i\u015f s\u00fcreci sa\u011flam\u0131\u015f bulunmaktad\u0131r. Ayr\u0131ca CDC sayesinden, veritaban\u0131nda istenilen tablonun tamam\u0131 yahut istek do\u011frultusunda bir k\u0131sm\u0131 \u00fczerinde yap\u0131lan de\u011fi\u015fiklikler takip edilerek s\u00fcre\u00e7teki verisel ak\u0131\u015f filtrelenebilmekte ve arzu edinildi\u011fi \u015fekilde y\u00f6netilebilmektedir.<\/p>\n

Debezium Nedir?<\/h4>\n
Giri\u015f paragraf\u0131nda bahsedildi\u011fi \u00fczere Debezium, veritaban\u0131 \u00fczerinde yap\u0131lan de\u011fi\u015fiklikleri yakalayarak bunlar\u0131 stream eden, open source ve distributed platform olan Change Data Capture(CDC) arac\u0131d\u0131r.<\/p>\n
Debezium, veritaban\u0131nda istenilen tabloda yahut o tablonun istenilen kolonlar\u0131nda olan t\u00fcm verisel de\u011fi\u015fiklikleri yakalayarak Kafka’ya aktar\u0131r. Bu de\u011fi\u015fiklikleri yakalama s\u00fcreci, Debezium’un \u00e7al\u0131\u015fmad\u0131\u011f\u0131 s\u00fcre\u00e7lerde de pasif bir \u015fekilde de ger\u00e7ekle\u015ftirilebilmektedir.<\/p>\n
Debezium, bir \u00fcstteki paragrafta bahsedildi\u011fi gibi veritaban\u0131nda olu\u015fan de\u011fi\u015fikliklere istinaden yakalad\u0131\u011f\u0131 verilerin dayan\u0131kl\u0131l\u0131\u011f\u0131 ve g\u00fcvenilirli\u011fi i\u00e7in Kafka ve Kafka Connect’ten yararlanmaktad\u0131r.<\/p>\n
Kafka Nedir?<\/h4>\n
Binlerce yaz\u0131l\u0131m projesi ve \u015firket taraf\u0131ndan kullan\u0131lan, publisher\/subscriber tabanl\u0131 olan open source distributed streaming platformudur. Temelde bilinmesi gereken Producer, Consumer, Broker ve Topic olmak \u00fczere d\u00f6rt terminolojik terimi mevcuttur;<\/p>\n
\n
Producer<\/em>, publisher rol\u00fcndedir. Subscriber’lara mesaj g\u00f6nderir.<\/li>\n
Consumer<\/em>, subscriber rol\u00fcndedir. Producer’lar\u0131n \u00fcretti\u011fi mesajlar\u0131 dinlemekte ve t\u00fcketmektedir.<\/li>\n
Topic<\/em>, Kafka’da mesajlar\u0131n tutuldu\u011fu yap\u0131d\u0131r.<\/li>\n
Broker<\/em>, Topic’leri bar\u0131nd\u0131ran sunuculard\u0131r.<\/li>\n<\/ul>\n
Debezium ile veritaban\u0131nda yakalanan de\u011fi\u015fikliklerin Kafka’ya aktar\u0131labilmesi i\u00e7in verilerin bunun anlayabilece\u011fi dilden bir formata d\u00f6n\u00fc\u015ft\u00fcr\u00fclmesi gerekmektedir. Bunun i\u00e7in Kafka Connect tool’unu kullan\u0131yor olaca\u011f\u0131z.<\/p>\n
Kafka Connect Nedir?<\/strong>
\nKafka’n\u0131n veritabanlar\u0131na, consule yahut eleasticsearch gibi yap\u0131lara ba\u011flanabilmesini ve metrikleri alabilmesini sa\u011flayan bir tool’dur. Hususi olarak i\u00e7eri\u011fimiz \u00e7er\u00e7evesinde de\u011ferlendirirsek e\u011fer Kafka ile Debezium aras\u0131nda veri ak\u0131\u015f\u0131n\u0131 sa\u011flayacak olan k\u00f6pr\u00fc g\u00f6revi g\u00f6ren yap\u0131lanmad\u0131r diyebiliriz.<\/p>\n
ZooKeeper Nedir?<\/h4>\n
Bir\u00e7ok yap\u0131land\u0131rma bilgisinin, distributed mimariler taraf\u0131ndan \u015fu veya bu \u015fekilde kullan\u0131ld\u0131\u011f\u0131n\u0131 biliyoruz. Tabi bu bilgilerin y\u00f6netimi manuel bir \u015fekilde yap\u0131ld\u0131\u011f\u0131 taktirde uygulanmalar\u0131n\u0131n zorlu\u011fu nedeniyle insani durumlardan kaynakl\u0131 hatalar ka\u00e7\u0131n\u0131lmaz olabilmektedir. Hadi diyelim do\u011fru bir \u015fekilde y\u00f6netildiler, bu sefer de uygulamalar da\u011f\u0131t\u0131ld\u0131\u011f\u0131nda y\u00f6netim karma\u015fas\u0131 gibi durumlarla kar\u015f\u0131la\u015f\u0131labilmektedir. \u0130\u015fte ZooKeeper, t\u00fcm bu riskleri g\u00f6zard\u0131 edip g\u00fcvenilir distributed yap\u0131land\u0131rma yapmam\u0131z\u0131 sa\u011flamaktad\u0131r.<\/p>\n
Haliyle, Kafka’n\u0131n yap\u0131land\u0131rma bilgilerini depolamak, adland\u0131rmak, senkronizasyonunu sa\u011flamak ve koordine etmek i\u00e7in ZooKeeper yaz\u0131l\u0131m\u0131n\u0131 kullan\u0131yor olaca\u011f\u0131z.<\/p>\n
Debezium entegrasyonlar\u0131nda ZooKeeper, temel kurulum d\u0131\u015f\u0131nda derinlemesine anla\u015f\u0131lmay\u0131 gerektirmemektedir.<\/em><\/p><\/blockquote>\n
\u015eimdi, temel enstr\u00fcmanlar\u0131 ve kullanaca\u011f\u0131m\u0131z teknolojileri teoride inceledi\u011fimize g\u00f6re art\u0131k pratikte Debezium’u inceleyebiliriz. Bunun i\u00e7in hedefledi\u011fim \u00fc\u00e7 farkl\u0131 veritaban\u0131 \u00fczerinden(PostgreSQL, MSSQL ve MongoDB)<\/span> \u00e7al\u0131\u015fma sergiliyor olaca\u011f\u0131z. Hadi buyrun, fazla vakit kaybetmeksizin ilk olarak PostgreSQL’den ba\u015flayal\u0131m.<\/p>\n
Debezium \u0130le PostgreSQL’de Change Data Capture(CDC) Nas\u0131l Yap\u0131l\u0131r?<\/h4>\n
\u0130lk olarak Debezium’u hangi veritaban\u0131yla kullan\u0131rsak kullanal\u0131m temelde Kafka ve Debezium \u00e7al\u0131\u015ft\u0131r\u0131lmal\u0131 ve ZooKeeper ile yap\u0131land\u0131rmalar ayarlanmal\u0131d\u0131r. T\u00fcm bu yaz\u0131l\u0131mlar\u0131m aya\u011fa kald\u0131r\u0131lmas\u0131 i\u00e7in Docker’dan istifade ediyor olaca\u011f\u0131z. \u015eimdi gelin ihtiyac\u0131m\u0131z olan ara\u00e7lar\u0131 h\u0131zl\u0131ca olu\u015fturup, aya\u011fa kald\u0131ral\u0131m.<\/p>\n
docker-compose-postgres.yaml<\/em><\/p>\n
\n
\r\nversion: '3.1'\r\nservices:\r\n zookeeper:\r\n image: debezium\/zookeeper\r\n ports:\r\n - "2181:2181"\r\n - "2888:2888"\r\n - "3888:3888"\r\n kafka:\r\n image: debezium\/kafka\r\n ports:\r\n - "9092:9092"\r\n - "29092:29092"\r\n depends_on:\r\n - zookeeper\r\n environment:\r\n - ZOOKEEPER_CONNECT=zookeeper:2181\r\n - KAFKA_ADVERTISED_LISTENERS=LISTENER_EXT:\/\/localhost:29092,LISTENER_INT:\/\/kafka:9092\r\n - KAFKA_LISTENER_SECURITY_PROTOCOL_MAP=LISTENER_INT:PLAINTEXT,LISTENER_EXT:PLAINTEXT\r\n - KAFKA_LISTENERS=LISTENER_INT:\/\/0.0.0.0:9092,LISTENER_EXT:\/\/0.0.0.0:29092\r\n - KAFKA_INTER_BROKER_LISTENER_NAME=LISTENER_INT\r\n postgres:\r\n image: debezium\/postgres\r\n ports:\r\n - 5432:5432\r\n environment:\r\n - POSTGRES_USER=postgres\r\n - POSTGRES_PASSWORD=123456\r\n - POSTGRES_DB=DebeziumDB\r\n connect:\r\n image: debezium\/connect\r\n ports:\r\n - 8083:8083\r\n environment:\r\n - BOOTSTRAP_SERVERS=kafka:9092\r\n - GROUP_ID=1\r\n - CONFIG_STORAGE_TOPIC=my_connect_configs\r\n - OFFSET_STORAGE_TOPIC=my_connect_offsets\r\n - STATUS_STORAGE_TOPIC=my_connect_statuses\r\n depends_on:\r\n - zookeeper\r\n - kafka\r\n<\/pre>\n<\/div>\n
Yukar\u0131daki docker compose dosyas\u0131na bakarsan\u0131z e\u011fer ihtiya\u00e7 do\u011frultusundaki ilgili t\u00fcm ara\u00e7lar\u0131 bar\u0131nd\u0131rmaktad\u0131r. T\u00fcm image’lar da debezium’lu versiyonlar\u0131 kullanmam\u0131z\u0131n nedeni debezium ile \u00e7al\u0131\u015fmalar\u0131 i\u00e7in gerekli ayarlar\u0131n\u0131n etkin gelmesindendir. PostgreSQL’de ise debezium’un sa\u011fl\u0131kl\u0131 \u00e7al\u0131\u015fabilmesi i\u00e7in wal_level<\/code> konfig\u00fcrasyonuna ‘logical’ de\u011feri verilmelidir. Bunun nedeni, PostgreSQL’de herhangi bir transaction y\u00fcr\u00fct\u00fcld\u00fc\u011f\u00fcnde \u00f6ncelikle bu i\u015flem Write-ahead logging(WAL)<\/em> ad\u0131 verilen yerde ger\u00e7ekle\u015ftirilmektedir. Debezium’da gerekli verileri WAL’dan toplayaca\u011f\u0131 i\u00e7in bu \u015fekilde ayarlanmas\u0131 gerekmektedir. Haliyle debezium\/postgres<\/code> bu konfig\u00fcrasyon temellerinde gelmektedir. \u015eimdi PostgreSQL sunucusunun yap\u0131land\u0131rmalar\u0131na bir g\u00f6z at\u0131p \u00f6yle devam edelim. \n<\/a><\/p>\n
Logical, WAL verilerini harici sistemlerin t\u00fcketebilmesi i\u00e7in gerekli izin konfig\u00fcrasyonunu sa\u011flar.<\/em><\/p><\/blockquote>\n
\u015eimdi bu docker compose dosyas\u0131n\u0131 \ndocker-compose -f docker-compose-postgres.yaml up<\/em><\/strong><\/code> \ntalimat\u0131 e\u015fli\u011finde \u00e7al\u0131\u015ft\u0131ral\u0131m. \n<\/a>T\u00fcm y\u00fcklemelerin bitip, container’lar\u0131n aya\u011fa kald\u0131r\u0131ld\u0131\u011f\u0131ndan emin olal\u0131m.<\/a><\/p>\n
\u015eimdi PostgreSQL sunucusunda yukar\u0131daki docker-compose-postgres.yaml<\/em> dosyas\u0131nda olu\u015fturulan ‘DebeziumDB’ veritaban\u0131 i\u00e7erisinde a\u015fa\u011f\u0131daki \u00f6rnek tabloyu ve \u015femay\u0131 olu\u015ftural\u0131m ve ard\u0131ndan \u00f6rnek bir ka\u00e7 veri insert edelim.<\/p>\n
\n\r\ncreate schema exampleschema\r\nset search_path TO exampleschema,public;\r\ncreate table exampleschema.exampletable (\r\n column1 int,\r\n column2 int,\r\n column3 varchar(150),\r\n primary key(column1)\r\n);\r\nalter table exampleschema.exampletable replica identity FULL;\r\ninsert into exampleschema.exampletable values (1000, 2000, 'example value');\r\ninsert into exampleschema.exampletable values (1001, 2001, 'example value 2');\r\n<\/pre>\n<\/div>\n\u015eimdi ise debezzium connector’\u0131n bizlere sunmu\u015f oldu\u011fu API arac\u0131l\u0131\u011f\u0131yla dinlenecek olan veritaban\u0131 ve tablo bilgilerinin verilmesi gerekmektedir. Bunun i\u00e7in a\u015fa\u011f\u0131daki gibi bir .json dosyas\u0131 olu\u015fturarak gerekli bilgileri i\u00e7erisinde tan\u0131mlamam\u0131z gerekmektedir.<\/p>\n register-postgres.json<\/em><\/p>\n\n\r\n{\r\n "name": "example-connector",\r\n "config": {\r\n "connector.class": "io.debezium.connector.postgresql.PostgresConnector",\r\n "tasks.max": "1",\r\n "database.hostname": "postgres",\r\n "database.port": "5432",\r\n "database.user": "postgres",\r\n "database.password": "123456",\r\n "database.dbname" : "DebeziumDB",\r\n "database.server.name": "exampleserver",\r\n "schema.include.list": "exampleschema",\r\n "table.whitelist": "exampleschema.exampletable"\r\n }\r\n}\r\n<\/pre>\n<\/div>\nBu bilgileri tan\u0131mlad\u0131ktan sonra curl<\/code> \u00fczerinden a\u015fa\u011f\u0131daki post iste\u011fini g\u00f6ndermemiz yeterlidir. \ncurl -i -X POST -H \"Accept:application\/json\" -H \"Content-Type:application\/json\" http:\/\/localhost:8083\/connectors\/ -d @register-postgres.json<\/em><\/strong><\/code><\/a>B\u00f6ylece bu son hamle e\u015fli\u011finde bir connector tan\u0131mlayarak debezium’u etkinle\u015ftirmi\u015f bulunmaktay\u0131z. Connector’\u0131 etkinle\u015ftirme neticesinde yukar\u0131daki g\u00f6rselde tek sat\u0131r halinde gelen result’\u0131 do\u011fru d\u00fczg\u00fcn g\u00f6rmek isteyenler i\u00e7in formatland\u0131r\u0131p a\u015fa\u011f\u0131ya b\u0131rak\u0131yorum;<\/p>\n \n\r\n{\r\n "name":"example-connector",\r\n "config":{\r\n "connector.class":"io.debezium.connector.postgresql.PostgresConnector",\r\n "tasks.max":"1",\r\n "database.hostname":"postgres",\r\n "database.port":"5432",\r\n "database.user":"postgres",\r\n "database.password":"123456",\r\n "database.dbname":"DebeziumDB",\r\n "database.server.name":"exampleserver",\r\n "schema.include.list":"exampleschema",\r\n "table.whitelist":"exampleschema.exampletable",\r\n "name":"example-connector"\r\n },\r\n "tasks":[\r\n \r\n ],\r\n "type":"source"\r\n}\r\n<\/pre>\n<\/div>\nVe \u015fimdi ise debezium connector ile takip edilen veritaban\u0131 sunucusundaki ilgili tablodaki veri de\u011fi\u015fikliklerinin izlemesini ger\u00e7ekle\u015ftirelim. Bunun i\u00e7in a\u015fa\u011f\u0131daki docker talimat\u0131n\u0131n verilmesi yeterlidir.<\/p>\ndocker-compose -f docker-compose-postgres.yaml exec kafka \/kafka\/bin\/kafka-console-consumer.sh --bootstrap-server kafka:9092 --from-beginning --property print.key=true --topic exampleserver.exampleschema.exampletable<\/em><\/strong><\/code> \n<\/a>Evet, g\u00f6r\u00fcld\u00fc\u011f\u00fc \u00fczere PostgreSQL veritaban\u0131ndaki yap\u0131lan de\u011fi\u015fiklikler yakalanm\u0131\u015f bulunmaktad\u0131r. Hatta debezium connector ba\u011flanmadan \u00f6nce girilen \u00f6rnek veriler bile \u00f6nceki paragraflarda bildirildi\u011fi \u00fczere debezium \u00e7al\u0131\u015ft\u0131r\u0131ld\u0131ktan sonra yakalanm\u0131\u015f ve elde edilmi\u015ftir. \u015eimdi gelen json verileri d\u00fczg\u00fcn formata getirip, inceleyelim.<\/p>\n \n\r\n{\r\n "schema":{\r\n "type":"struct",\r\n "fields":[\r\n {\r\n "type":"int32",\r\n "optional":false,\r\n "field":"column1"\r\n }\r\n ],\r\n "optional":false,\r\n "name":"exampleserver.exampleschema.exampletable.Key"\r\n },\r\n "payload":{\r\n "column1":1000\r\n }\r\n}{\r\n "schema":{\r\n "type":"struct",\r\n "fields":[\r\n {\r\n "type":"struct",\r\n "fields":[\r\n {\r\n "type":"int32",\r\n "optional":false,\r\n "field":"column1"\r\n },\r\n {\r\n "type":"int32",\r\n "optional":true,\r\n "field":"column2"\r\n },\r\n {\r\n "type":"string",\r\n "optional":true,\r\n "field":"column3"\r\n }\r\n ],\r\n "optional":true,\r\n "name":"exampleserver.exampleschema.exampletable.Value",\r\n "field":"before"\r\n },\r\n {\r\n "type":"struct",\r\n "fields":[\r\n {\r\n "type":"int32",\r\n "optional":false,\r\n "field":"column1"\r\n },\r\n {\r\n "type":"int32",\r\n "optional":true,\r\n "field":"column2"\r\n },\r\n {\r\n "type":"string",\r\n "optional":true,\r\n "field":"column3"\r\n }\r\n ],\r\n "optional":true,\r\n "name":"exampleserver.exampleschema.exampletable.Value",\r\n "field":"after"\r\n },\r\n {\r\n "type":"struct",\r\n "fields":[\r\n {\r\n "type":"string",\r\n "optional":false,\r\n "field":"version"\r\n },\r\n {\r\n "type":"string",\r\n "optional":false,\r\n "field":"connector"\r\n },\r\n {\r\n "type":"string",\r\n "optional":false,\r\n "field":"name"\r\n },\r\n {\r\n "type":"int64",\r\n "optional":false,\r\n "field":"ts_ms"\r\n },\r\n {\r\n "type":"string",\r\n "optional":true,\r\n "name":"io.debezium.data.Enum",\r\n "version":1,\r\n "parameters":{\r\n "allowed":"true,last,false,incremental"\r\n },\r\n "default":"false",\r\n "field":"snapshot"\r\n },\r\n {\r\n "type":"string",\r\n "optional":false,\r\n "field":"db"\r\n },\r\n {\r\n "type":"string",\r\n "optional":true,\r\n "field":"sequence"\r\n },\r\n {\r\n "type":"string",\r\n "optional":false,\r\n "field":"schema"\r\n },\r\n {\r\n "type":"string",\r\n "optional":false,\r\n "field":"table"\r\n },\r\n {\r\n "type":"int64",\r\n "optional":true,\r\n "field":"txId"\r\n },\r\n {\r\n "type":"int64",\r\n "optional":true,\r\n "field":"lsn"\r\n },\r\n {\r\n "type":"int64",\r\n "optional":true,\r\n "field":"xmin"\r\n }\r\n ],\r\n "optional":false,\r\n "name":"io.debezium.connector.postgresql.Source",\r\n "field":"source"\r\n },\r\n {\r\n "type":"string",\r\n "optional":false,\r\n "field":"op"\r\n },\r\n {\r\n "type":"int64",\r\n "optional":true,\r\n "field":"ts_ms"\r\n },\r\n {\r\n "type":"struct",\r\n "fields":[\r\n {\r\n "type":"string",\r\n "optional":false,\r\n "field":"id"\r\n },\r\n {\r\n "type":"int64",\r\n "optional":false,\r\n "field":"total_order"\r\n },\r\n {\r\n "type":"int64",\r\n "optional":false,\r\n "field":"data_collection_order"\r\n }\r\n ],\r\n "optional":true,\r\n "field":"transaction"\r\n }\r\n ],\r\n "optional":false,\r\n "name":"exampleserver.exampleschema.exampletable.Envelope"\r\n },\r\n "payload":{\r\n "before":null,\r\n "after":{\r\n "column1":1000,\r\n "column2":2000,\r\n "column3":"example value"\r\n },\r\n "source":{\r\n "version":"1.8.1.Final",\r\n "connector":"postgresql",\r\n "name":"exampleserver",\r\n "ts_ms":1647763176365,\r\n "snapshot":"true",\r\n "db":"DebeziumDB",\r\n "sequence":"[null,\\"23723656\\"]",\r\n "schema":"exampleschema",\r\n "table":"exampletable",\r\n "txId":557,\r\n "lsn":23723656,\r\n "xmin":null\r\n },\r\n "op":"r",\r\n "ts_ms":1647763176368,\r\n "transaction":null\r\n }\r\n}\r\n<\/pre>\n<\/div>\nYukar\u0131daki json datay\u0131 incelerseniz e\u011fer 177 ile 200. sat\u0131r aral\u0131\u011f\u0131nda de\u011fi\u015fiklik olan verileri i\u00e7eren ‘payload’ alan\u0131 mevcuttur. Yakalanan verilerin hangi aksiyona tabii tutuldu\u011funu 198. sat\u0131rdaki ‘op’ alan\u0131ndaki c<\/code>, u<\/code>, d<\/code> ve r<\/code> de\u011ferleri ifade etmektedir.<\/p>\n\n\n\n\n\nc<\/strong><\/td>\n Yakalanan verinin eklendi\u011fini ifade eder.<\/td>\n<\/tr>\n u<\/strong><\/td>\n Yakalanan verinin g\u00fcncellendi\u011fini ifade eder.<\/td>\n<\/tr>\n d<\/strong><\/td>\n Yakalanan verinin silindi\u011fini ifade eder.<\/td>\n<\/tr>\n r<\/strong><\/td>\n Yakalanan verinin zaten var oldu\u011funu ifade eder.<\/td>\n<\/tr>\n<\/table>\nPeki hoca, gelen json data’da ki di\u011fer alanlar neyin nesi?<\/em><\/strong> diye sorarsan\u0131z e\u011fer \u015f\u00f6yle cevaplayal\u0131m;<\/p>\n \nschema<\/em>; takip edilen tablonun yap\u0131s\u0131 ve kolonlarla ilgili bilgiler ta\u015f\u0131yan k\u0131s\u0131md\u0131r.<\/li>\n payload<\/em>; yukar\u0131da g\u00f6r\u00fcd\u00fc\u011f\u00fcm\u00fcz \u00fczere de\u011fi\u015fiklikler neticesinde akan verilere dair bilgi veren k\u0131s\u0131md\u0131r.<\/li>\n before<\/em>; de\u011fi\u015fiklik olan verilerin, de\u011fi\u015fiklik olmadan \u00f6nceki hallerini ta\u015f\u0131yan k\u0131s\u0131md\u0131r. Insert haricinde update ve delete sorgular\u0131nda dolu gelmektedir.<\/li>\n after<\/em>; de\u011fi\u015fiklik olan verilerdeki son de\u011ferleri veren k\u0131s\u0131md\u0131r. Delete hari\u00e7 di\u011fer sorgularda dolu gelmektedir.<\/li>\n<\/ul>\n\u015eimdi test ama\u00e7l\u0131 a\u015fa\u011f\u0131daki veritaban\u0131 sorgular\u0131n\u0131 \u00e7al\u0131\u015ft\u0131ral\u0131m.<\/p>\n \n\r\nupdate exampleschema.exampletable\r\nset\r\ncolumn1 = 5000,\r\ncolumn2 = 6000,\r\ncolumn3 = 'updated example value'\r\nwhere column1 = 1000\r\n<\/pre>\n<\/div>\n<\/a>G\u00f6r\u00fcld\u00fc\u011f\u00fc \u00fczere \u00e7al\u0131\u015ft\u0131r\u0131lan sorgu update sorgusu ise veritaban\u0131 a\u00e7\u0131s\u0131ndan bu bir delete ve insert i\u015flemi oldu\u011fu i\u00e7in \u00f6nce ‘op’ de\u011feri ‘d’ sonra da ‘c’ olan iki kay\u0131t yakalanmaktad\u0131r. Haliyle ‘before’ ve ‘after’ alanlar\u0131ndan update yap\u0131lan sorguya dair verileri rahatl\u0131kla yakalayabilmekteyiz.<\/p>\n \n\r\ndelete from exampleschema.exampletable\r\nwhere column1 = 5000\r\n<\/pre>\n<\/div>\n<\/a>Delete sorgusunda da ayn\u0131 mant\u0131k silinecek veriye dair \u00f6nceki kay\u0131tlar elde edilebilmektedir.<\/p>\n B\u00f6ylece PostgreSQL ile CDC \u00e7al\u0131\u015fmas\u0131n\u0131 ger\u00e7ekle\u015ftirmi\u015f olduk. Mevzu bahis, Debezium ile yakalanan verilerin Kafka \u00fczerinden .NET Core ortam\u0131nda okunmas\u0131na gelmeden \u00f6nce MSSQL ve MongoDB \u00fczerinden de CDC entegrasyonunu ele alal\u0131m. Nihayetinde t\u00fcm \u00f6rneklendirmeler Kafka kullanaca\u011f\u0131ndan dolay\u0131 i\u00e7eri\u011fimizin sonlar\u0131na do\u011fru .NET Core ortam\u0131nda m\u00fc\u015fterek bir uygulama olu\u015fturarak konuyu tek elden \u00f6rneklendirebiliriz.<\/p>\n Debezium \u0130le MSSQL’de Change Data Capture(CDC) Nas\u0131l Yap\u0131l\u0131r?<\/h4>\n\u0130lk olarak MSSQL veritaban\u0131yla birlikte ilgili image’lar\u0131 aya\u011fa kald\u0131rabilmek i\u00e7in a\u015fa\u011f\u0131daki docker compose dosyas\u0131n\u0131 olu\u015ftural\u0131m.<\/p>\n docker-compose-sqlserver.yaml<\/em><\/p>\n\n\r\nversion: '3.1'\r\nservices:\r\n zookeeper:\r\n image: debezium\/zookeeper\r\n ports:\r\n - "2181:2181"\r\n - "2888:2888"\r\n - "3888:3888"\r\n kafka:\r\n image: debezium\/kafka\r\n ports:\r\n - "9092:9092"\r\n - "29092:29092"\r\n depends_on:\r\n - zookeeper\r\n environment:\r\n - ZOOKEEPER_CONNECT=zookeeper:2181\r\n - KAFKA_ADVERTISED_LISTENERS=LISTENER_EXT:\/\/localhost:29092,LISTENER_INT:\/\/kafka:9092\r\n - KAFKA_LISTENER_SECURITY_PROTOCOL_MAP=LISTENER_INT:PLAINTEXT,LISTENER_EXT:PLAINTEXT\r\n - KAFKA_LISTENERS=LISTENER_INT:\/\/0.0.0.0:9092,LISTENER_EXT:\/\/0.0.0.0:29092\r\n - KAFKA_INTER_BROKER_LISTENER_NAME=LISTENER_INT\r\n sqlserver:\r\n image: mcr.microsoft.com\/mssql\/server\r\n ports:\r\n - 1433:1433\r\n environment:\r\n - ACCEPT_EULA=Y\r\n - MSSQL_PID=Standard\r\n - SA_PASSWORD=1q2w3e4r!^+\r\n - MSSQL_AGENT_ENABLED=true\r\n connect:\r\n image: debezium\/connect\r\n ports:\r\n - 8083:8083\r\n environment:\r\n - BOOTSTRAP_SERVERS=kafka:9092\r\n - GROUP_ID=1\r\n - CONFIG_STORAGE_TOPIC=my_connect_configs\r\n - OFFSET_STORAGE_TOPIC=my_connect_offsets\r\n - STATUS_STORAGE_TOPIC=my_connect_statuses\r\n depends_on:\r\n - zookeeper\r\n - kafka\r\n<\/pre>\n<\/div>\nArd\u0131ndan a\u015fa\u011f\u0131daki talimat e\u015fli\u011finde ilgili dosyay\u0131 \u00e7al\u0131\u015ft\u0131ral\u0131m. \ndocker-compose -f docker-compose-sqlserver.yaml up<\/em><\/strong><\/code><\/a>\u015eimdi SQL Server’da \u00f6rnek bir veritaban\u0131 ve tablo olu\u015ftural\u0131m.<\/p>\n \n\r\ncreate database DebeziumDB\r\ngo\r\nuse DebeziumDB\r\ngo\r\ncreate schema exampleschema\r\ngo\r\ncreate table exampleschema.exampletable\r\n(\r\n\tcolumn1 int primary key identity(1,1),\r\n\tcolumn2 int,\r\n\tcolumn3 nvarchar(max)\r\n)\r\n<\/pre>\n<\/div>\nAkabinde olu\u015fturulan veritaban\u0131 i\u00e7erisinde verisel de\u011fi\u015fiklikleri yakalayabilmek i\u00e7in gerekli CDC aktifle\u015ftirme konfig\u00fcrasyonunu ger\u00e7ekle\u015ftirelim.<\/p>\n\n\r\nuse DebeziumDB\r\ngo\r\nexec sys.sp_cdc_enable_db\r\n<\/pre>\n<\/div>\nBu veritaban\u0131 seviyesinde bir CDC aktifle\u015ftirmeyken,<\/p>\n\n\r\nuse DebeziumDB\r\ngo\r\nexec sys.sp_cdc_enable_table \r\n@source_schema = N'exampleschema', \r\n@source_name = N'exampletable', \r\n@role_name = NULL, \r\n@filegroup_name = N'', \r\n@supports_net_changes = 0 \r\n<\/pre>\n<\/div>\nbu ise tablo seviyesinde CDC aktifle\u015ftirmesidir.<\/p>\n Bu i\u015flemlerden sonra s\u0131ra debezium i\u00e7in bir connector tan\u0131mlamas\u0131nda bulunmaya gelmi\u015ftir. Bunun i\u00e7in a\u015fa\u011f\u0131daki register konfig\u00fcrasyonlar\u0131n\u0131 bar\u0131nd\u0131ran .json dosyas\u0131n\u0131 curl \u00fczerinden verilen talimat e\u015fli\u011finde post edilmesi yeterlidir.<\/p>\n register-sqlserver.json<\/em><\/p>\n\n\r\n{\r\n "name": "example-connector",\r\n "config": {\r\n "connector.class": "io.debezium.connector.sqlserver.SqlServerConnector",\r\n "tasks.max": "1",\r\n "database.hostname": "sqlserver",\r\n "database.port": "1433",\r\n "database.user": "sa",\r\n "database.password": "1q2w3e4r!^+",\r\n "database.dbname": "DebeziumDB",\r\n "database.server.name": "exampleserver",\r\n "schema.include.list": "exampleschema",\r\n "table.whitelist": "exampleschema.exampletable",\r\n "database.history.kafka.bootstrap.servers": "kafka:9092",\r\n "database.history.kafka.topic": "dbhistory.fullfillment"\r\n }\r\n}\r\n<\/pre>\n<\/div>\ncurl -i -X POST -H \"Accept:application\/json\" -H \"Content-Type:application\/json\" http:\/\/localhost:8083\/connectors\/ -d @register-sqlserver.json<\/em><\/strong><\/code> \n<\/a>Yukar\u0131daki g\u00f6rseli incelerseniz e\u011fer connector olu\u015fturma neticesinde result’unu bizlere d\u00f6nd\u00fcrm\u00fc\u015f vaziyettedir. Result’un do\u011fru formatland\u0131r\u0131lm\u0131\u015f halini bir \u00f6nceki PostgreSQL \u00f6rneklendirmesinde sundu\u011fum i\u00e7in burada l\u00fczum g\u00f6rmemekteyim.<\/p>\n Art\u0131k debezium ile ba\u011flant\u0131s\u0131 kurulan SQL Server’\u0131n \u00fczerinde yap\u0131lan de\u011fi\u015fikliklerin yakalan\u0131p, yakalanmad\u0131\u011f\u0131n\u0131 a\u015fa\u011f\u0131daki talimat e\u015fli\u011finde takip edebiliriz. \ndocker-compose -f docker-compose-sqlserver.yaml exec kafka \/kafka\/bin\/kafka-console-consumer.sh --bootstrap-server kafka:9092 --from-beginning --property print.key=true --topic exampleserver.exampleschema.exampletable<\/em><\/strong><\/code> \n<\/a><\/p>\n Evet… SQL Server ile de CDC \u00e7al\u0131\u015fmas\u0131n\u0131 tamamlam\u0131\u015f bulunmaktay\u0131z. \u015eimdi s\u0131rada NoSQL yakla\u015f\u0131m\u0131na sahip olan MongoDB var \ud83d\ude42<\/p>\n Debezium \u0130le MongoDB’de Change Data Capture(CDC) Nas\u0131l Yap\u0131l\u0131r?<\/h4>\nMongoDB i\u00e7in \u00f6rneklendirmede biraz h\u0131zl\u0131 ilerliyor olaca\u011f\u0131z. \u015eimdi a\u015fa\u011f\u0131daki docker compose i\u00e7eri\u011fini verilen talimat e\u015fli\u011finde \u00e7al\u0131\u015ft\u0131rarak ba\u015flayal\u0131m.<\/p>\n\n\r\nversion: '2'\r\nservices:\r\n zookeeper:\r\n image: quay.io\/debezium\/zookeeper\r\n ports:\r\n - 2181:2181\r\n - 2888:2888\r\n - 3888:3888\r\n kafka:\r\n image: quay.io\/debezium\/kafka\r\n ports:\r\n - 9092:9092\r\n links:\r\n - zookeeper\r\n environment:\r\n - ZOOKEEPER_CONNECT=zookeeper:2181\r\n mongodb:\r\n image: quay.io\/debezium\/example-mongodb\r\n hostname: mongodb\r\n ports:\r\n - 27017:27017\r\n environment:\r\n - MONGODB_USER=debezium\r\n - MONGODB_PASSWORD=dbz\r\n connect:\r\n image: quay.io\/debezium\/connect\r\n ports:\r\n - 8083:8083\r\n links:\r\n - kafka\r\n - mongodb\r\n environment:\r\n - BOOTSTRAP_SERVERS=kafka:9092\r\n - GROUP_ID=1\r\n - CONFIG_STORAGE_TOPIC=my_connect_configs\r\n - OFFSET_STORAGE_TOPIC=my_connect_offsets\r\n - STATUS_STORAGE_TOPIC=my_connect_statuses\r\n<\/pre>\n<\/div>\ndocker-compose -f docker-compose-mongodb.yaml up<\/em><\/strong><\/code><\/p>\n Akabinde MongoDB replika setini ba\u015flatal\u0131m ve test verileri ekleyelim.<\/p>\n\n\r\ndocker-compose -f docker-compose-mongodb.yaml exec mongodb bash -c '\/usr\/local\/bin\/init-inventory.sh'\r\n<\/pre>\n<\/div>\nArd\u0131ndan a\u015fa\u011f\u0131daki i\u00e7eri\u011fe sahip olan connector’u verilen talimat e\u015fli\u011finde ba\u015flatal\u0131m.<\/p>\n\n\r\n{\r\n "name": "inventory-connector",\r\n "config": {\r\n "connector.class" : "io.debezium.connector.mongodb.MongoDbConnector",\r\n "tasks.max" : "1",\r\n "mongodb.hosts" : "rs0\/mongodb:27017",\r\n "mongodb.name" : "dbserver1",\r\n "mongodb.user" : "debezium",\r\n "mongodb.password" : "dbz",\r\n "database.include.list" : "inventory",\r\n "database.history.kafka.bootstrap.servers" : "kafka:9092"\r\n }\r\n}\r\n<\/pre>\n<\/div>\ncurl -i -X POST -H \"Accept:application\/json\" -H \"Content-Type:application\/json\" http:\/\/localhost:8083\/connectors\/ -d @register-mongodb.json<\/em><\/strong><\/code><\/p>\nYap\u0131lan bu \u00e7al\u0131\u015fmalar kafidir. example-mongodb<\/em> image’i sayesinde \u00f6rnek bir veritaban\u0131 olu\u015fturulmu\u015f vaziyettedir. Haliyle bu veritaban\u0131 i\u00e7erisindeki verisel de\u011fi\u015fiklikleri yakalayan kafka’ya ba\u011flanarak verileri t\u00fcketebiliriz. Bunun i\u00e7in a\u015fa\u011f\u0131daki talimat\u0131n verilmesi yeterlidir. \ndocker-compose -f docker-compose-mongodb.yaml exec kafka \/kafka\/bin\/kafka-console-consumer.sh --bootstrap-server kafka:9092 --from-beginning --property print.key=true --topic dbserver1.inventory.customers<\/em><\/strong><\/code> \n<\/a>Vee i\u015fte MongoDB’dende yap\u0131lan verisel de\u011fi\u015fikliklerin yakaland\u0131\u011f\u0131n\u0131 hep birlikte g\u00f6r\u00fcyoruz \ud83d\ude42<\/p>\n \u015eimdi veritabanlar\u0131nda yap\u0131lan de\u011fi\u015fikliklerin CDC ile yakalan\u0131p kafka’ya g\u00f6nderilen datalar\u0131n\u0131 C# \u00fczerinden nas\u0131l okuyabilece\u011fimizi basit olarak ele alman\u0131n vakti geldi.<\/p>\n Debezium \u0130le PostgreSQL, MSSQL veya MongoDB’den Change Data Capture(CDC) \u0130le Yakalanan Verilerin C# Taraf\u0131ndan Okunmas\u0131<\/h4>\nBunun i\u00e7in a\u015fa\u011f\u0131daki gibi Confluent.Kafka<\/a> k\u00fct\u00fcphanesini kullanarak basit bir consumer olu\u015fturabiliriz.<\/p>\n \n\r\nusing Confluent.Kafka;\r\n\r\nConsumerConfig config = new()\r\n{\r\n GroupId = "exampleserver.exampleschema.exampletable",\r\n BootstrapServers = "localhost:29092",\r\n AutoOffsetReset = AutoOffsetReset.Earliest,\r\n\r\n};\r\n\r\nusing IConsumer<Ignore, string> consumer = new ConsumerBuilder<Ignore, string>(config).Build();\r\nconsumer.Subscribe("exampleserver.exampleschema.exampletable");\r\n\r\nCancellationTokenSource source = new CancellationTokenSource();\r\nConsole.CancelKeyPress += (_, e) =>\r\n{\r\n e.Cancel = true;\r\n source.Cancel();\r\n};\r\nwhile (true)\r\n{\r\n ConsumeResult<Ignore, string> result = consumer.Consume(source.Token);\r\n Console.WriteLine($"Topic Name : {result.TopicPartitionOffset}");\r\n Console.WriteLine($"Message : {result.Value}");\r\n}\r\n<\/pre>\n<\/div>\nEkstra Bilgiler<\/h4>\nMisal, Kafka’da herhangi bir topic olu\u015fturuldu mu? ya da yukar\u0131 sat\u0131rlarda olu\u015fturdu\u011fumuz connector’lar hala ayakta m\u0131? Bunun i\u00e7in a\u015fa\u011f\u0131daki talimatlar e\u015fli\u011finde sorgulama yapabiliyoruz…<\/p>\n\n\n\nMevcut connectorlar<\/td>\n :<\/td>\n http:\/\/localhost:8083\/connectors\/<\/code><\/td>\n<\/tr>\n Topic list<\/td>\n :<\/td>\n docker-compose -f docker-compose-postgres.yaml exec kafka \/kafka\/bin\/kafka-topics.sh --list --bootstrap-server kafka:9092<\/code> \nya da \nbin\/kafka-topics.sh --list --bootstrap-server kafka:9092<\/code><\/td>\n<\/tr>\n<\/table>\n\u0130lgilenenlerin faydalanmas\u0131 dile\u011fiyle… \nSonraki yaz\u0131lar\u0131mda g\u00f6r\u00fc\u015fmek \u00fczere… \n\u0130yi \u00e7al\u0131\u015fmalar…<\/p>\n Not : \u00d6rnek dosyalar\u0131 a\u015fa\u011f\u0131daki linklere t\u0131klayarak indirebilirsiniz. \nMongoDB<\/a>, SQLServer<\/a>, Postgres<\/a>, DebeziumExample<\/a><\/p>\n","protected":false},"excerpt":{"rendered":" Merhaba, Bu i\u00e7eri\u011fimizde birden \u00e7ok veri kayna\u011f\u0131nda ger\u00e7ek zamanl\u0131 olu\u015fan de\u011fi\u015fikliklerin yakalanmas\u0131n\u0131 ve stream edilmesini sa\u011flayan, open source ve distributed bir ara\u00e7 olan Debezium CDC(Change Data Capture)’yi inceliyor olaca\u011f\u0131z. CDC(Change Data Capture) Nedir? Veritaban\u0131nda...<\/p>\n","protected":false},"author":1,"featured_media":24620,"comment_status":"open","ping_status":"open","sticky":false,"template":"","format":"standard","meta":{"footnotes":""},"categories":[4536],"tags":[4540,4541,4539,4542,4544,4546,4545,4543,4551,4549,4550,4548,4547],"class_list":["post-24601","post","type-post","status-publish","format-standard","has-post-thumbnail","hentry","category-debezium","tag-cdk","tag-change-data-capture","tag-debezium","tag-debezium-nedir","tag-kafka","tag-kafka-connect","tag-kafka-connect-nedir","tag-kafka-nedir","tag-wal","tag-wal2json","tag-write-ahead-logging","tag-zookeeper","tag-zookeeper-nedir"],"amp_enabled":true,"_links":{"self":[{"href":"https:\/\/www.gencayyildiz.com\/blog\/wp-json\/wp\/v2\/posts\/24601","targetHints":{"allow":["GET"]}}],"collection":[{"href":"https:\/\/www.gencayyildiz.com\/blog\/wp-json\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/www.gencayyildiz.com\/blog\/wp-json\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/www.gencayyildiz.com\/blog\/wp-json\/wp\/v2\/users\/1"}],"replies":[{"embeddable":true,"href":"https:\/\/www.gencayyildiz.com\/blog\/wp-json\/wp\/v2\/comments?post=24601"}],"version-history":[{"count":133,"href":"https:\/\/www.gencayyildiz.com\/blog\/wp-json\/wp\/v2\/posts\/24601\/revisions"}],"predecessor-version":[{"id":24759,"href":"https:\/\/www.gencayyildiz.com\/blog\/wp-json\/wp\/v2\/posts\/24601\/revisions\/24759"}],"wp:featuredmedia":[{"embeddable":true,"href":"https:\/\/www.gencayyildiz.com\/blog\/wp-json\/wp\/v2\/media\/24620"}],"wp:attachment":[{"href":"https:\/\/www.gencayyildiz.com\/blog\/wp-json\/wp\/v2\/media?parent=24601"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/www.gencayyildiz.com\/blog\/wp-json\/wp\/v2\/categories?post=24601"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/www.gencayyildiz.com\/blog\/wp-json\/wp\/v2\/tags?post=24601"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}

c<\/strong><\/td>\n	Yakalanan verinin eklendi\u011fini ifade eder.<\/td>\n<\/tr>\n
u<\/strong><\/td>\n	Yakalanan verinin g\u00fcncellendi\u011fini ifade eder.<\/td>\n<\/tr>\n
d<\/strong><\/td>\n	Yakalanan verinin silindi\u011fini ifade eder.<\/td>\n<\/tr>\n
r<\/strong><\/td>\n	Yakalanan verinin zaten var oldu\u011funu ifade eder.<\/td>\n<\/tr>\n<\/table>\n *Peki hoca, gelen json data’da ki di\u011fer alanlar neyin nesi?<\/em><\/strong> diye sorarsan\u0131z e\u011fer \u015f\u00f6yle cevaplayal\u0131m;<\/p>\n* \n schema<\/em>; takip edilen tablonun yap\u0131s\u0131 ve kolonlarla ilgili bilgiler ta\u015f\u0131yan k\u0131s\u0131md\u0131r.<\/li>\n payload<\/em>; yukar\u0131da g\u00f6r\u00fcd\u00fc\u011f\u00fcm\u00fcz \u00fczere de\u011fi\u015fiklikler neticesinde akan verilere dair bilgi veren k\u0131s\u0131md\u0131r.<\/li>\n before<\/em>; de\u011fi\u015fiklik olan verilerin, de\u011fi\u015fiklik olmadan \u00f6nceki hallerini ta\u015f\u0131yan k\u0131s\u0131md\u0131r. Insert haricinde update ve delete sorgular\u0131nda dolu gelmektedir.<\/li>\n after<\/em>; de\u011fi\u015fiklik olan verilerdeki son de\u011ferleri veren k\u0131s\u0131md\u0131r. Delete hari\u00e7 di\u011fer sorgularda dolu gelmektedir.<\/li>\n<\/ul>\n \u015eimdi test ama\u00e7l\u0131 a\u015fa\u011f\u0131daki veritaban\u0131 sorgular\u0131n\u0131 \u00e7al\u0131\u015ft\u0131ral\u0131m.<\/p>\n \n \r\nupdate exampleschema.exampletable\r\nset\r\ncolumn1 = 5000,\r\ncolumn2 = 6000,\r\ncolumn3 = 'updated example value'\r\nwhere column1 = 1000\r\n<\/pre>\n<\/div>\n $\"Debezium(CDC)$ <\/a>G\u00f6r\u00fcld\u00fc\u011f\u00fc \u00fczere \u00e7al\u0131\u015ft\u0131r\u0131lan sorgu update sorgusu ise veritaban\u0131 a\u00e7\u0131s\u0131ndan bu bir delete ve insert i\u015flemi oldu\u011fu i\u00e7in \u00f6nce ‘op’ de\u011feri ‘d’ sonra da ‘c’ olan iki kay\u0131t yakalanmaktad\u0131r. Haliyle ‘before’ ve ‘after’ alanlar\u0131ndan update yap\u0131lan sorguya dair verileri rahatl\u0131kla yakalayabilmekteyiz.<\/p>\n \n \r\ndelete from exampleschema.exampletable\r\nwhere column1 = 5000\r\n<\/pre>\n<\/div>\n $\"Debezium(CDC)$ <\/a>Delete sorgusunda da ayn\u0131 mant\u0131k silinecek veriye dair \u00f6nceki kay\u0131tlar elde edilebilmektedir.<\/p>\n B\u00f6ylece PostgreSQL ile CDC \u00e7al\u0131\u015fmas\u0131n\u0131 ger\u00e7ekle\u015ftirmi\u015f olduk. Mevzu bahis, Debezium ile yakalanan verilerin Kafka \u00fczerinden .NET Core ortam\u0131nda okunmas\u0131na gelmeden \u00f6nce MSSQL ve MongoDB \u00fczerinden de CDC entegrasyonunu ele alal\u0131m. Nihayetinde t\u00fcm \u00f6rneklendirmeler Kafka kullanaca\u011f\u0131ndan dolay\u0131 i\u00e7eri\u011fimizin sonlar\u0131na do\u011fru .NET Core ortam\u0131nda m\u00fc\u015fterek bir uygulama olu\u015fturarak konuyu tek elden \u00f6rneklendirebiliriz.<\/p>\n Debezium \u0130le MSSQL’de Change Data Capture(CDC) Nas\u0131l Yap\u0131l\u0131r?<\/h4>\n \u0130lk olarak MSSQL veritaban\u0131yla birlikte ilgili image’lar\u0131 aya\u011fa kald\u0131rabilmek i\u00e7in a\u015fa\u011f\u0131daki docker compose dosyas\u0131n\u0131 olu\u015ftural\u0131m.<\/p>\n docker-compose-sqlserver.yaml<\/em><\/p>\n \n \r\nversion: '3.1'\r\nservices:\r\n zookeeper:\r\n image: debezium\/zookeeper\r\n ports:\r\n - "2181:2181"\r\n - "2888:2888"\r\n - "3888:3888"\r\n kafka:\r\n image: debezium\/kafka\r\n ports:\r\n - "9092:9092"\r\n - "29092:29092"\r\n depends_on:\r\n - zookeeper\r\n environment:\r\n - ZOOKEEPER_CONNECT=zookeeper:2181\r\n - KAFKA_ADVERTISED_LISTENERS=LISTENER_EXT:\/\/localhost:29092,LISTENER_INT:\/\/kafka:9092\r\n - KAFKA_LISTENER_SECURITY_PROTOCOL_MAP=LISTENER_INT:PLAINTEXT,LISTENER_EXT:PLAINTEXT\r\n - KAFKA_LISTENERS=LISTENER_INT:\/\/0.0.0.0:9092,LISTENER_EXT:\/\/0.0.0.0:29092\r\n - KAFKA_INTER_BROKER_LISTENER_NAME=LISTENER_INT\r\n sqlserver:\r\n image: mcr.microsoft.com\/mssql\/server\r\n ports:\r\n - 1433:1433\r\n environment:\r\n - ACCEPT_EULA=Y\r\n - MSSQL_PID=Standard\r\n - SA_PASSWORD=1q2w3e4r!^+\r\n - MSSQL_AGENT_ENABLED=true\r\n connect:\r\n image: debezium\/connect\r\n ports:\r\n - 8083:8083\r\n environment:\r\n - BOOTSTRAP_SERVERS=kafka:9092\r\n - GROUP_ID=1\r\n - CONFIG_STORAGE_TOPIC=my_connect_configs\r\n - OFFSET_STORAGE_TOPIC=my_connect_offsets\r\n - STATUS_STORAGE_TOPIC=my_connect_statuses\r\n depends_on:\r\n - zookeeper\r\n - kafka\r\n<\/pre>\n<\/div>\n Ard\u0131ndan a\u015fa\u011f\u0131daki talimat e\u015fli\u011finde ilgili dosyay\u0131 \u00e7al\u0131\u015ft\u0131ral\u0131m. \n`docker-compose -f docker-compose-sqlserver.yaml up<\/em><\/strong><\/code><\/a>\u015eimdi SQL Server’da \u00f6rnek bir veritaban\u0131 ve tablo olu\u015ftural\u0131m.<\/p>\n` \n\r\ncreate database DebeziumDB\r\ngo\r\nuse DebeziumDB\r\ngo\r\ncreate schema exampleschema\r\ngo\r\ncreate table exampleschema.exampletable\r\n(\r\n\tcolumn1 int primary key identity(1,1),\r\n\tcolumn2 int,\r\n\tcolumn3 nvarchar(max)\r\n)\r\n<\/pre>\n<\/div>\nAkabinde olu\u015fturulan veritaban\u0131 i\u00e7erisinde verisel de\u011fi\u015fiklikleri yakalayabilmek i\u00e7in gerekli CDC aktifle\u015ftirme konfig\u00fcrasyonunu ger\u00e7ekle\u015ftirelim.<\/p>\n\n\r\nuse DebeziumDB\r\ngo\r\nexec sys.sp_cdc_enable_db\r\n<\/pre>\n<\/div>\nBu veritaban\u0131 seviyesinde bir CDC aktifle\u015ftirmeyken,<\/p>\n\n\r\nuse DebeziumDB\r\ngo\r\nexec sys.sp_cdc_enable_table \r\n@source_schema = N'exampleschema', \r\n@source_name = N'exampletable', \r\n@role_name = NULL, \r\n@filegroup_name = N'', \r\n@supports_net_changes = 0 \r\n<\/pre>\n<\/div>\nbu ise tablo seviyesinde CDC aktifle\u015ftirmesidir.<\/p>\n Bu i\u015flemlerden sonra s\u0131ra debezium i\u00e7in bir connector tan\u0131mlamas\u0131nda bulunmaya gelmi\u015ftir. Bunun i\u00e7in a\u015fa\u011f\u0131daki register konfig\u00fcrasyonlar\u0131n\u0131 bar\u0131nd\u0131ran .json dosyas\u0131n\u0131 curl \u00fczerinden verilen talimat e\u015fli\u011finde post edilmesi yeterlidir.<\/p>\n register-sqlserver.json<\/em><\/p>\n\n\r\n{\r\n "name": "example-connector",\r\n "config": {\r\n "connector.class": "io.debezium.connector.sqlserver.SqlServerConnector",\r\n "tasks.max": "1",\r\n "database.hostname": "sqlserver",\r\n "database.port": "1433",\r\n "database.user": "sa",\r\n "database.password": "1q2w3e4r!^+",\r\n "database.dbname": "DebeziumDB",\r\n "database.server.name": "exampleserver",\r\n "schema.include.list": "exampleschema",\r\n "table.whitelist": "exampleschema.exampletable",\r\n "database.history.kafka.bootstrap.servers": "kafka:9092",\r\n "database.history.kafka.topic": "dbhistory.fullfillment"\r\n }\r\n}\r\n<\/pre>\n<\/div>\ncurl -i -X POST -H \"Accept:application\/json\" -H \"Content-Type:application\/json\" http:\/\/localhost:8083\/connectors\/ -d @register-sqlserver.json<\/em><\/strong><\/code> \n<\/a>Yukar\u0131daki g\u00f6rseli incelerseniz e\u011fer connector olu\u015fturma neticesinde result’unu bizlere d\u00f6nd\u00fcrm\u00fc\u015f vaziyettedir. Result’un do\u011fru formatland\u0131r\u0131lm\u0131\u015f halini bir \u00f6nceki PostgreSQL \u00f6rneklendirmesinde sundu\u011fum i\u00e7in burada l\u00fczum g\u00f6rmemekteyim.<\/p>\n Art\u0131k debezium ile ba\u011flant\u0131s\u0131 kurulan SQL Server’\u0131n \u00fczerinde yap\u0131lan de\u011fi\u015fikliklerin yakalan\u0131p, yakalanmad\u0131\u011f\u0131n\u0131 a\u015fa\u011f\u0131daki talimat e\u015fli\u011finde takip edebiliriz. \ndocker-compose -f docker-compose-sqlserver.yaml exec kafka \/kafka\/bin\/kafka-console-consumer.sh --bootstrap-server kafka:9092 --from-beginning --property print.key=true --topic exampleserver.exampleschema.exampletable<\/em><\/strong><\/code> \n<\/a><\/p>\n Evet… SQL Server ile de CDC \u00e7al\u0131\u015fmas\u0131n\u0131 tamamlam\u0131\u015f bulunmaktay\u0131z. \u015eimdi s\u0131rada NoSQL yakla\u015f\u0131m\u0131na sahip olan MongoDB var \ud83d\ude42<\/p>\n Debezium \u0130le MongoDB’de Change Data Capture(CDC) Nas\u0131l Yap\u0131l\u0131r?<\/h4>\nMongoDB i\u00e7in \u00f6rneklendirmede biraz h\u0131zl\u0131 ilerliyor olaca\u011f\u0131z. \u015eimdi a\u015fa\u011f\u0131daki docker compose i\u00e7eri\u011fini verilen talimat e\u015fli\u011finde \u00e7al\u0131\u015ft\u0131rarak ba\u015flayal\u0131m.<\/p>\n\n\r\nversion: '2'\r\nservices:\r\n zookeeper:\r\n image: quay.io\/debezium\/zookeeper\r\n ports:\r\n - 2181:2181\r\n - 2888:2888\r\n - 3888:3888\r\n kafka:\r\n image: quay.io\/debezium\/kafka\r\n ports:\r\n - 9092:9092\r\n links:\r\n - zookeeper\r\n environment:\r\n - ZOOKEEPER_CONNECT=zookeeper:2181\r\n mongodb:\r\n image: quay.io\/debezium\/example-mongodb\r\n hostname: mongodb\r\n ports:\r\n - 27017:27017\r\n environment:\r\n - MONGODB_USER=debezium\r\n - MONGODB_PASSWORD=dbz\r\n connect:\r\n image: quay.io\/debezium\/connect\r\n ports:\r\n - 8083:8083\r\n links:\r\n - kafka\r\n - mongodb\r\n environment:\r\n - BOOTSTRAP_SERVERS=kafka:9092\r\n - GROUP_ID=1\r\n - CONFIG_STORAGE_TOPIC=my_connect_configs\r\n - OFFSET_STORAGE_TOPIC=my_connect_offsets\r\n - STATUS_STORAGE_TOPIC=my_connect_statuses\r\n<\/pre>\n<\/div>\ndocker-compose -f docker-compose-mongodb.yaml up<\/em><\/strong><\/code><\/p>\n Akabinde MongoDB replika setini ba\u015flatal\u0131m ve test verileri ekleyelim.<\/p>\n\n\r\ndocker-compose -f docker-compose-mongodb.yaml exec mongodb bash -c '\/usr\/local\/bin\/init-inventory.sh'\r\n<\/pre>\n<\/div>\nArd\u0131ndan a\u015fa\u011f\u0131daki i\u00e7eri\u011fe sahip olan connector’u verilen talimat e\u015fli\u011finde ba\u015flatal\u0131m.<\/p>\n\n\r\n{\r\n "name": "inventory-connector",\r\n "config": {\r\n "connector.class" : "io.debezium.connector.mongodb.MongoDbConnector",\r\n "tasks.max" : "1",\r\n "mongodb.hosts" : "rs0\/mongodb:27017",\r\n "mongodb.name" : "dbserver1",\r\n "mongodb.user" : "debezium",\r\n "mongodb.password" : "dbz",\r\n "database.include.list" : "inventory",\r\n "database.history.kafka.bootstrap.servers" : "kafka:9092"\r\n }\r\n}\r\n<\/pre>\n<\/div>\ncurl -i -X POST -H \"Accept:application\/json\" -H \"Content-Type:application\/json\" http:\/\/localhost:8083\/connectors\/ -d @register-mongodb.json<\/em><\/strong><\/code><\/p>\nYap\u0131lan bu \u00e7al\u0131\u015fmalar kafidir. example-mongodb<\/em> image’i sayesinde \u00f6rnek bir veritaban\u0131 olu\u015fturulmu\u015f vaziyettedir. Haliyle bu veritaban\u0131 i\u00e7erisindeki verisel de\u011fi\u015fiklikleri yakalayan kafka’ya ba\u011flanarak verileri t\u00fcketebiliriz. Bunun i\u00e7in a\u015fa\u011f\u0131daki talimat\u0131n verilmesi yeterlidir. \ndocker-compose -f docker-compose-mongodb.yaml exec kafka \/kafka\/bin\/kafka-console-consumer.sh --bootstrap-server kafka:9092 --from-beginning --property print.key=true --topic dbserver1.inventory.customers<\/em><\/strong><\/code> \n<\/a>Vee i\u015fte MongoDB’dende yap\u0131lan verisel de\u011fi\u015fikliklerin yakaland\u0131\u011f\u0131n\u0131 hep birlikte g\u00f6r\u00fcyoruz \ud83d\ude42<\/p>\n \u015eimdi veritabanlar\u0131nda yap\u0131lan de\u011fi\u015fikliklerin CDC ile yakalan\u0131p kafka’ya g\u00f6nderilen datalar\u0131n\u0131 C# \u00fczerinden nas\u0131l okuyabilece\u011fimizi basit olarak ele alman\u0131n vakti geldi.<\/p>\n Debezium \u0130le PostgreSQL, MSSQL veya MongoDB’den Change Data Capture(CDC) \u0130le Yakalanan Verilerin C# Taraf\u0131ndan Okunmas\u0131<\/h4>\nBunun i\u00e7in a\u015fa\u011f\u0131daki gibi Confluent.Kafka<\/a> k\u00fct\u00fcphanesini kullanarak basit bir consumer olu\u015fturabiliriz.<\/p>\n \n\r\nusing Confluent.Kafka;\r\n\r\nConsumerConfig config = new()\r\n{\r\n GroupId = "exampleserver.exampleschema.exampletable",\r\n BootstrapServers = "localhost:29092",\r\n AutoOffsetReset = AutoOffsetReset.Earliest,\r\n\r\n};\r\n\r\nusing IConsumer<Ignore, string> consumer = new ConsumerBuilder<Ignore, string>(config).Build();\r\nconsumer.Subscribe("exampleserver.exampleschema.exampletable");\r\n\r\nCancellationTokenSource source = new CancellationTokenSource();\r\nConsole.CancelKeyPress += (_, e) =>\r\n{\r\n e.Cancel = true;\r\n source.Cancel();\r\n};\r\nwhile (true)\r\n{\r\n ConsumeResult<Ignore, string> result = consumer.Consume(source.Token);\r\n Console.WriteLine($"Topic Name : {result.TopicPartitionOffset}");\r\n Console.WriteLine($"Message : {result.Value}");\r\n}\r\n<\/pre>\n<\/div>\nEkstra Bilgiler<\/h4>\nMisal, Kafka’da herhangi bir topic olu\u015fturuldu mu? ya da yukar\u0131 sat\u0131rlarda olu\u015fturdu\u011fumuz connector’lar hala ayakta m\u0131? Bunun i\u00e7in a\u015fa\u011f\u0131daki talimatlar e\u015fli\u011finde sorgulama yapabiliyoruz…<\/p>\n\n\n\nMevcut connectorlar<\/td>\n :<\/td>\n http:\/\/localhost:8083\/connectors\/<\/code><\/td>\n<\/tr>\n Topic list<\/td>\n :<\/td>\n docker-compose -f docker-compose-postgres.yaml exec kafka \/kafka\/bin\/kafka-topics.sh --list --bootstrap-server kafka:9092<\/code> \nya da \nbin\/kafka-topics.sh --list --bootstrap-server kafka:9092<\/code><\/td>\n<\/tr>\n<\/table>\n\u0130lgilenenlerin faydalanmas\u0131 dile\u011fiyle… \nSonraki yaz\u0131lar\u0131mda g\u00f6r\u00fc\u015fmek \u00fczere… \n\u0130yi \u00e7al\u0131\u015fmalar…<\/p>\n Not : \u00d6rnek dosyalar\u0131 a\u015fa\u011f\u0131daki linklere t\u0131klayarak indirebilirsiniz. \nMongoDB<\/a>, SQLServer<\/a>, Postgres<\/a>, DebeziumExample<\/a><\/p>\n","protected":false},"excerpt":{"rendered":" Merhaba, Bu i\u00e7eri\u011fimizde birden \u00e7ok veri kayna\u011f\u0131nda ger\u00e7ek zamanl\u0131 olu\u015fan de\u011fi\u015fikliklerin yakalanmas\u0131n\u0131 ve stream edilmesini sa\u011flayan, open source ve distributed bir ara\u00e7 olan Debezium CDC(Change Data Capture)’yi inceliyor olaca\u011f\u0131z. CDC(Change Data Capture) Nedir? Veritaban\u0131nda...<\/p>\n","protected":false},"author":1,"featured_media":24620,"comment_status":"open","ping_status":"open","sticky":false,"template":"","format":"standard","meta":{"footnotes":""},"categories":[4536],"tags":[4540,4541,4539,4542,4544,4546,4545,4543,4551,4549,4550,4548,4547],"class_list":["post-24601","post","type-post","status-publish","format-standard","has-post-thumbnail","hentry","category-debezium","tag-cdk","tag-change-data-capture","tag-debezium","tag-debezium-nedir","tag-kafka","tag-kafka-connect","tag-kafka-connect-nedir","tag-kafka-nedir","tag-wal","tag-wal2json","tag-write-ahead-logging","tag-zookeeper","tag-zookeeper-nedir"],"amp_enabled":true,"_links":{"self":[{"href":"https:\/\/www.gencayyildiz.com\/blog\/wp-json\/wp\/v2\/posts\/24601","targetHints":{"allow":["GET"]}}],"collection":[{"href":"https:\/\/www.gencayyildiz.com\/blog\/wp-json\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/www.gencayyildiz.com\/blog\/wp-json\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/www.gencayyildiz.com\/blog\/wp-json\/wp\/v2\/users\/1"}],"replies":[{"embeddable":true,"href":"https:\/\/www.gencayyildiz.com\/blog\/wp-json\/wp\/v2\/comments?post=24601"}],"version-history":[{"count":133,"href":"https:\/\/www.gencayyildiz.com\/blog\/wp-json\/wp\/v2\/posts\/24601\/revisions"}],"predecessor-version":[{"id":24759,"href":"https:\/\/www.gencayyildiz.com\/blog\/wp-json\/wp\/v2\/posts\/24601\/revisions\/24759"}],"wp:featuredmedia":[{"embeddable":true,"href":"https:\/\/www.gencayyildiz.com\/blog\/wp-json\/wp\/v2\/media\/24620"}],"wp:attachment":[{"href":"https:\/\/www.gencayyildiz.com\/blog\/wp-json\/wp\/v2\/media?parent=24601"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/www.gencayyildiz.com\/blog\/wp-json\/wp\/v2\/categories?post=24601"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/www.gencayyildiz.com\/blog\/wp-json\/wp\/v2\/tags?post=24601"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}