Add : Example for MySQL to Kafka events

examples/mysql_to_kafka.py

@@ -0,0 +1,83 @@
+#!/usr/bin/env python
+
+#
+# Output Kafka events to the console from MySQL replication stream
+#
+
+from pymysqlreplication import BinLogStreamReader
+from pymysqlreplication.row_event import (
+    DeleteRowsEvent,
+    UpdateRowsEvent,
+    WriteRowsEvent,
+)
+
+from kafka.admin import KafkaAdminClient, NewTopic
+from kafka import KafkaConsumer, KafkaProducer
+
+MYSQL_SETTINGS = {"host": "127.0.0.1", "port": 3306, "user": "root", "passwd": ""}
+
+
+def create_kafka_topic(topic_name, num_partitions=1, replication_factor=1):
+    admin_client = KafkaAdminClient(bootstrap_servers="127.0.0.1:9092")
+
+    topic = NewTopic(
+        name=topic_name,
+        num_partitions=num_partitions,
+        replication_factor=replication_factor,
+    )
+
+    admin_client.create_topics(new_topics=[topic])
+
+
+def main():
+    global message_body
+    producer = KafkaProducer(
+        bootstrap_servers="127.0.0.1:9092",
+        value_serializer=lambda v: str(v).encode("utf-8"),
+    )
+
+    stream = BinLogStreamReader(
+        connection_settings=MYSQL_SETTINGS,
+        server_id=3,
+        only_events=[DeleteRowsEvent, UpdateRowsEvent, WriteRowsEvent],
+    )
+
+    for binlogevent in stream:
+        for row in binlogevent.rows:
+            if isinstance(binlogevent, DeleteRowsEvent):
+                topic = "deleted"
+                message_body = row["values"].items()
+
+            elif isinstance(binlogevent, UpdateRowsEvent):
+                topic = "updated"
+                message_body = row["after_values"].items()
+
+            elif isinstance(binlogevent, WriteRowsEvent):
+                topic = "created"
+                message_body = row["values"].items()
+
+            producer.send(topic, key=None, value=dict(message_body))
+
+    consumer = KafkaConsumer(
+        "deleted",
+        "updated",
+        "created",
+        bootstrap_servers="127.0.0.1:9092",
+        value_deserializer=lambda x: x.decode("utf-8"),
+        auto_offset_reset="earliest",
+        group_id="1",
+    )
+
+    for message in consumer:
+        print(f'Topic: "{message.topic}", Value: "{message.value}"')
+
+    stream.close()
+    producer.close()
+    consumer.close()
+
+
+if __name__ == "__main__":
+    create_kafka_topic("deleted", num_partitions=3, replication_factor=1)
+    create_kafka_topic("created", num_partitions=3, replication_factor=1)
+    create_kafka_topic("updated", num_partitions=3, replication_factor=1)
+    main()