Home

login

linkedin

Help

About me

Github

Courses

Medium

%23%23%20Why%0A%0AIn%20the%20era%20of%20modern%20data-driven%20applications%2C%20the%20significance%20of%20data%20streaming%20is%20continuously%20on%20the%20rise.%20This%20upward%20trend%20can%20be%20attributed%20to%20various%20factors%2C%20including%20real-time%20data%20processing%2C%20scalability%2C%20event%20sourcing%2C%20data%20transformation%2C%20and%20complex%20event%20processing.%0A%0AIn%20this%20article%2C%20we%20will%20explore%20an%20example%20where%20data%20streaming%20is%20essential%20and%20demonstrate%20how%20to%20implement%20it%20using%20Debezium%2C%20Apache%20Pulsar%2C%20and%20Go.%0A%0A%23%23%20Use%20Case%0A%0AImagine%20that%20we%20are%20developing%20an%20online%20bookstore%20similar%20to%20Amazon.%20Our%20tech%20stack%20includes%20PostgreSQL%20as%20the%20primary%20data%20storage%2C%20Golang%20for%20building%20the%20backend%20API%2C%20and%20Elasticsearch%20for%20efficient%20data%20searching.%20The%20initial%20challenge%20we%20encounter%20is%20how%20to%20efficiently%20synchronize%20data%20from%20PostgreSQL%20to%20Elasticsearch.%0A%0A%23%23%23%20Requirements%0A%0A**Functional%3A**%0A-%20Efficient%20data%20synchronization%20from%20PostgreSQL%20to%20Elasticsearch.%0A-%20Near-real-time%20data%20processing%20for%20changes.%0A%0A**Non-Functional%3A**%0A-%20Scalability%20in%20data%20synchronization%20to%20handle%20increasing%20data%20volumes.%0A-%20Low-latency%20data%20synchronization%20for%20real-time%20responsiveness.%0A-%20Secure%20data%20transfer%20and%20storage%20during%20synchronization.%0A-%20Fault-tolerant%20data%20synchronization%20to%20handle%20potential%20failures.%0A-%20Monitoring%20and%20logging%20for%20tracking%20synchronization%20performance%20and%20troubleshooting.%0A-%20Minimal%20impact%20on%20the%20performance%20of%20the%20source%20PostgreSQL%20database.%0A-%20Compliance%20with%20data%20privacy%20and%20security%20regulations%20in%20data%20handling%20and%20storage.%0A%0A%23%23%23%20Key%20Considerations%0A%0ATo%20accomplish%20this%2C%20we%20require%20a%20system%20capable%20of%20effectively%20capturing%20data%20from%20PostgreSQL%20while%20also%20being%20scalable.%20An%20established%20solution%20that%20fulfills%20these%20criteria%20is%20Debezium.%20Once%20Debezium%20captures%20the%20data%2C%20it%20promptly%20forwards%20it%20to%20a%20message%20queue%20to%20alleviate%20the%20workload%20and%20speed%20up%20processing.%20Now%2C%20we%20require%20a%20message%20queue%20that%20can%20effectively%20store%20the%20data%2C%20is%20scalable%2C%20operates%20with%20minimal%20latency%2C%20and%20ensures%20fault%20tolerance.%20In%20this%20solution%20we%20will%20be%20using%20Apache%20Pulsar%20which%20meets%20the%20above%20requirements.%0A%0A%23%23%20High%20Level%20Overview%0A%0A!%5B%5D(https%3A%2F%2Fcdn.shopemaa.cloud%2F4f877f28-3e2a-47a4-a195-2689d1789ed3%2Fmedia%2F0ae7a6f8-1482-4bec-9c5c-0ba898f5b491-U2NyZWVuc2hvdCAyMDIzLTA5LTA1IGF0IDQuNTcuMDYgUE0%3D.png)%0A%0A%23%23%20Implementation%0A-%20Data%20alterations%20in%20a%20table%20will%20be%20directed%20to%20a%20dedicated%20topic%2C%20ensuring%20data%20isolation%20for%20each%20table%20and%20facilitating%20easy%20scalability%20as%20data%20volume%20increases%20for%20a%20particular%20table.%0A-%20In%20Apache%20Pulsar%2C%20topic%20partitions%20can%20be%20dynamically%20expanded%20or%20reduced%2C%20guaranteeing%20automatic%20partition%20scaling%20in%20response%20to%20data%20growth.%0A-%20In%20terms%20of%20implementation%2C%20a%20consumer%20will%20consume%20data%20from%20a%20designated%20topic%2C%20ensuring%20the%20flexibility%20to%20scale%20consumers%20as%20required.%20To%20maintain%20the%20expected%20performance%2C%20it's%20essential%20to%20scale%20consumers%20as%20data%20volume%20increases.%0A%0A%23%23%20Setup%0A%0A**Debezium%20using%20docker-compose**%0A%0A%2F%2F%20application.properties%0A%60%60%60yml%0Adebezium.sink.type%3Dpulsar%0Adebezium.sink.pulsar.client.serviceUrl%3D%7Bpulsar_server_url%7D%0Adebezium.sink.pulsar.client.authParams%3Dtoken%3A%7Bjwt_auth_token%7D%0Adebezium.sink.pulsar.client.authPluginClassName%3Dorg.apache.pulsar.client.impl.auth.AuthenticationToken%0Adebezium.sink.pulsar.tenant%3D%7Btenant_name%7D%0Adebezium.source.connector.class%3Dio.debezium.connector.postgresql.PostgresConnector%0Adebezium.source.database.hostname%3D%7Bpostgresql_host%7D%0Adebezium.source.database.port%3D%7Bpostgresql_port%7D%0Adebezium.source.database.user%3D%7Bpostgresql_username%7D%0Adebezium.source.database.password%3D%7Bpostgresql_password%7D%0Adebezium.source.database.dbname%3D%7Bpostgresql_database_name%7D%0Adebezium.source.plugin.name%3Dpgoutput%0Adebezium.source.table.include.list%3D%7Bpostgresql_tables_comma_seperated%7D%0Adebezium.source.database.server.name%3D%7Bgive_a_random_name%7D%0Adebezium.format.key%3Djson%0Adebezium.format.value%3Djson%0Adebezium.source.topic.prefix%3D%7Bpulsar_topic_prefix%7D%0Adebezium.source.publication.autocreate.mode%3Dfiltered%0A%60%60%60%0A%0A%2F%2F%20docker-compose.yml%0A%60%60%60yml%0Aversion%3A%20'3.1'%0A%0Aservices%3A%0A%09debezium_server%3A%0A%09%09image%3A%20docker%20pull%20debezium%2Fserver%3A2.3.3.Final%0A%09%09volumes%3A%0A%09%09%09-%20%22.%2Fapplication.properties%3A%2Fdebezium%2Fconf%2Fapplication.properties%22%0A%60%60%60%0A%0APlease%20verify%20that%20the%20database%20has%20logical%20Write-Ahead%20Logging%20(WAL)%20enabled.%20When%20we%20execute%20the%20docker-compose%20command%2C%20it%20will%20retrieve%20data%20from%20the%20database%20and%20send%20it%20to%20Pulsar.%20At%20this%20stage%2C%20our%20data%20is%20now%20within%20Pulsar%2C%20and%20it's%20time%20to%20transfer%20the%20data%20from%20Pulsar%20to%20Elasticsearch.%0A%0A**Install%20go%20pulsar%20client**%0A%0A%60%60%60bash%0Ago%20get%20-u%20github.com%2Fapache%2Fpulsar-client-go%2Fpulsar%0A%60%60%60%0A%0A**Initialize%20go%20client%2C**%0A%0A%60%60%60go%0Aclient%2C%20err%20%3A%3D%20pulsar.NewClient(pulsar.ClientOptions%7B%0A%09%09URL%3A%20%7B%20ServiceUrl%20%7D%2C%0A%09%09OperationTimeout%3A%20%2030%20*%20time.Second%2C%0A%09%09ConnectionTimeout%3A%2030%20*%20time.Second%2C%0A%09%09Authentication%3A%20%7Bpulsar_jwt_token%7D%2C%20%2F%2F%20format%20%3D%20token%3Ajwt_token%0A%09%7D)%0A%09if%20err%20!%3D%20nil%20%7B%0A%09%09panic(err)%0A%09%7D%0A%60%60%60%0A%0A**Subscribe%20to%20topic**%0A%0A%60%60%60go%0Aconsumer%2C%20err%20%3A%3D%20pulsarCon.Subscribe(pulsar.ConsumerOptions%7B%0A%09%09%09%09Topics%3A%20%5B%5Dstring%7Btopic%7D%2C%0A%09%09%09%09Type%3A%20pulsar.Shared%2C%0A%09%09%09%09SubscriptionName%3A%20%7Bsubscription_name%7D%2C%0A%09%09%09%09SubscriptionInitialPosition%3A%20pulsar.SubscriptionPositionLatest%2C%0A%7D)%0Aif%20err%20!%3D%20nil%20%7B%0A%20%20panic(err)%0A%7D%0A%0Afor%20cm%20%3A%3D%20range%20consumer.Chan()%20%7B%0A%20%20%20msg%20%3A%3D%20cm.Message%0A%20%20%20%2F%2F%20transform%20data%20as%20needed%20and%20send%20to%20elastic%0A%09%20%2F%2F%20after%20processing%20execute%20cm.Ack(msg)%20for%20acknowledgement%0A%09%20%2F%2F%20and%20cm.Nack(msg)%20to%20re-process%20at%20a%20later%20time%0A%7D%0A%60%60%60%0A%0ANote%20that%20%60pulsar.Shared%60%20allows%20multiple%20consumers%20to%20consume%20from%20same%20topic.%0A%0ANow%2C%20we%20should%20have%20our%20data%20in%20Elasticsearch.%20To%20ensure%20the%20integrity%20of%20the%20Elasticsearch%20index%2C%20we%20can%20query%20it%20and%20validate%20some%20data%20to%20confirm%20that%20all%20the%20expected%20data%20is%20present.%0A%0ATo%20learn%20more%20about%20%5BApache%20Pulsar%5D(https%3A%2F%2Fpulsar.apache.org%2F)%20and%20%5BDebezium%5D(https%3A%2F%2Fdebezium.io%2F)%20go%20over%20their%20documentation.%0A%0A%0A%23%23%23%23%23%20Your%20caffeine-powered%20contribution%20keeps%20the%20code%20flowing%20and%20the%20ideas%20brewing!%20%E2%98%95%F0%9F%92%A1%0A%5B!%5BBuy%20Me%20A%20Coffee%5D(https%3A%2F%2Fimg.buymeacoffee.com%2Fbutton-api%2F%3Ftext%3DBuy%2520me%2520a%2520coffee%26emoji%3D%26slug%3Dsakib.ninja%26button_colour%3DFFDD00%26font_colour%3D000000%26font_family%3DBree%26outline_colour%3D000000%26coffee_colour%3Dffffff)%5D(https%3A%2F%2Fwww.buymeacoffee.com%2Fsakib.ninja)

Building data pipeline using Debezium & Apache Pulsar

Bangladesh

Learn with Sakib