Logstash sample

julien-duponchelle · julien-duponchelle · commit c6d69ff072f0 · 2013-06-26T22:44:19.000+02:00
diff --git a/examples/logstash/logstash-simple.conf b/examples/logstash/logstash-simple.conf
@@ -0,0 +1,11 @@
+input {
+    stdin {
+        type => "mysql_event"
+        format => "json_event"
+        debug => true
+    }
+}
+output { 
+  stdout { debug => true debug_format => "json"}
+  elasticsearch { embedded => true }
+}
diff --git a/examples/logstash/mysql_to_logstash.py b/examples/logstash/mysql_to_logstash.py
@@ -0,0 +1,57 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+
+#
+# Output logstash events to the console
+#
+# You can pipe it to logstash like this:
+# python examples/logstash/mysql_to_logstash.py | java -jar logstash-1.1.13-flatjar.jar  agent -f examples/logstash/logstash-simple.conf
+
+import json
+import time
+import sys
+
+from pymysqlreplication import BinLogStreamReader
+from pymysqlreplication.row_event import (
+    DeleteRowsEvent,
+    UpdateRowsEvent,
+    WriteRowsEvent,
+)
+
+MYSQL_SETTINGS = {
+    "host": "127.0.0.1",
+    "port": 3306,
+    "user": "root",
+    "passwd": ""
+}
+
+
+def main():
+    stream = BinLogStreamReader(
+        connection_settings=MYSQL_SETTINGS,
+        only_events=[DeleteRowsEvent, WriteRowsEvent, UpdateRowsEvent])
+
+    for binlogevent in stream:
+        for row in binlogevent.rows:
+            event = {}
+            event["schema"] = binlogevent.schema
+            event["table"] = binlogevent.table
+
+            if isinstance(binlogevent, DeleteRowsEvent):
+                event["action"] = "delete"
+                event = dict(event.items() + row["values"].items())
+            elif isinstance(binlogevent, UpdateRowsEvent):
+                event["action"] = "update"
+                event = dict(event.items() + row["after_values"].items())
+            elif isinstance(binlogevent, WriteRowsEvent):
+                event["action"] = "insert"
+                event = dict(event.items() + row["values"].items())
+            print json.dumps(event)
+            sys.stdout.flush()
+
+
+    stream.close()
+
+
+if __name__ == "__main__":
+    main()
