Debezium.

Debezium Overview

Debezium is an open-source distributed platform for change data capture (CDC). It streams real-time changes from databases to messaging systems like Kafka. Here's a breakdown:

---

Key Points

1. Core Purpose: Captures database changes (inserts, updates, deletes) for downstream consumers in near real-time, ensuring eventual consistency.

2. Supported Databases: Includes PostgreSQL, MySQL, MongoDB, Oracle, SQL Server, and Db2. Connector support varies.

3. Architecture: Debezium connectors run as Kafka Connect tasks, translating database events into Kafka topics.

4. Event Format: Events contain before, after, and metadata fields in JSON/Avro, representing old/new states and details like operation type.

5. Use Cases: Ideal for microservices communication, audit trails, real-time analytics, and database migrations.

---

Example Codes

1. Configuring a PostgreSQL Connector

{
  "name": "postgres-connector",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "database.hostname": "localhost",
    "database.port": "5432",
    "database.user": "dbuser",
    "database.password": "dbpassword",
    "database.dbname": "mydb",
    "database.server.name": "postgres_server",
    "table.include.list": "public.my_table"
  }
}

2. Consuming Change Events from Kafka

from kafka import KafkaConsumer
import json

consumer = KafkaConsumer(
    'postgres_server.public.my_table',
    bootstrap_servers='localhost:9092',
    value_deserializer=lambda m: json.loads(m.decode('utf-8'))
)

for message in consumer:
    print(f"Change Event: {message.value}")

3. Monitoring Kafka Connectors

curl -X GET http://localhost:8083/connectors/postgres-connector/status

4. Event Processing in Real-Time

from kafka import KafkaConsumer

consumer = KafkaConsumer('postgres_server.public.my_table', bootstrap_servers='localhost:9092')
for event in consumer:
    print(f"Operation: {event.value['op']}, Data: {event.value['after']}")

5. Transforming Events

{
  "transforms": "unwrap",
  "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
  "transforms.unwrap.drop.tombstones": "true"
}

---