First commit

Author: scottpersinger

.env

@@ -0,0 +1 @@
+RDS_URL=mysql://admin:[email protected]:3306/magento

Gemfile

@@ -0,0 +1 @@
+gem 'poseidon'

README.md

@@ -0,0 +1,27 @@
+# Mysql-kafka-replication
+
+Demonstration of using Mysql replication protocol to stream Mysql changes into Kafka queues, then
+read from those queues and process Mysql changes in a Node.js script.
+
+## Installation
+
+    pip install -r requirements.txt
+    npm install
+
+Oh, and go setup [Kafka](http://kafka.apache.org/) and get it running on `localhost:9092`.
+
+## Running
+
+Set `RDS_URL` in your environment, or add to a local `.env` file and run with `foreman`.
+
+First, run the Mysql replication listener:
+
+    python mysql-replicate-to-kafka.py
+
+and then run the Node.js app:
+
+    node consumer.js
+
+Now go make some changes to your database and see them appear in the console of the node app.
+
+

config.js

@@ -0,0 +1,5 @@
+module.exports = {
+	databases: {
+		db1: 'postgres://localhost/herokuconnect'
+	}
+}

consumer.js

@@ -0,0 +1,45 @@
+var kafka = require('kafka-node'),
+    Consumer = kafka.Consumer,
+    knex = require('knex'),
+	glob = require("glob"),
+	path = require("path"),
+    client = new kafka.Client();
+
+var config = require('./config.js');
+for (var key in config.databases) {
+	config[key] = knex({client:'pg', connection:config.databases[key], debug:true});
+}
+
+var handlers = {};
+var topics = [];
+
+glob("./scripts/*.js", function(er, files) {
+	console.log(files);
+	files.forEach(function(f) {
+		var mod = require(f);
+		if (typeof(mod) == "object") {
+			for (var topic in mod) {
+				topics.push({topic: topic, partition:0});
+				handlers[topic] = mod[topic];
+			}
+		} else {
+			var topic = path.basename(f, '.js');
+			topics.push({topic: topic, partition:0});
+			handlers[topic] = mod;
+		}
+	});
+
+	console.log("Topics ", topics);
+
+	var consumer = new Consumer(client, topics);
+
+	consumer.on('message', function(message) {
+		try {
+			handlers[message.topic](config, JSON.parse(message.value));
+		} catch (e) {
+			console.log(e);
+		}
+	});
+});
+
+

consumer.rb

@@ -0,0 +1,11 @@
+require 'poseidon'
+
+consumer = Poseidon::PartitionConsumer.new("my_test_consumer", "localhost", 9092,
+										"herokutest.tutors", 0, :earliest_offset)
+
+loop do
+	messages = consumer.fetch
+	messages.each do |m|
+		puts m.value
+	end
+end		

mysql-replicate-to-kafka.py

@@ -0,0 +1,83 @@
+import os
+import json
+import time
+import urlparse
+from datetime import datetime, date
+from decimal import Decimal
+
+from pymysqlreplication import BinLogStreamReader
+from pymysqlreplication.row_event import BINLOG
+from kafka import SimpleProducer, KafkaClient
+from kafka.common import LeaderNotAvailableError
+
+
+def json_serial(obj):
+    """JSON serializer for objects not serializable by default json code"""
+
+    if isinstance(obj, (datetime, date)):
+        serial = obj.isoformat()
+        return serial
+    if isinstance(obj, Decimal):
+    	return float(obj)
+    else:
+    	print "Type '{}' for '{}' not serializable".format(obj.__class__, obj)
+    	return None
+
+def build_message(binlog_evt):
+	schema = {'table': getattr(binlog_evt, 'schema', '') + "." + getattr(binlog_evt, 'table', '')};
+
+	if binlog_evt.event_type == BINLOG.WRITE_ROWS_EVENT_V2:
+		# Insert
+		return {'event':'INSERT', 'headers':schema, 'data':binlog_evt.rows[0]['values']}
+
+	elif binlog_evt.event_type == BINLOG.UPDATE_ROWS_EVENT_V2:
+		# Update
+		return {'event':'UPDATE', 'headers':schema, 'data':binlog_evt.rows[0]['after_values']}
+	elif binlog_evt.event_type == BINLOG.DELETE_ROWS_EVENT_V2:
+		# Delete
+		return {'event':'DELETE', 'headers':schema, 'data':binlog_evt.rows[0]['values']}
+
+	else:
+		return None
+
+
+kafka = KafkaClient("localhost:9092")
+
+producer = SimpleProducer(kafka)
+producer.send_messages("test", "test msg")
+
+# To wait for acknowledgements
+# ACK_AFTER_LOCAL_WRITE : server will wait till the data is written to
+#                         a local log before sending response
+# ACK_AFTER_CLUSTER_COMMIT : server will block until the message is committed
+#                            by all in sync replicas before sending a response
+producer = SimpleProducer(kafka, async=False,
+                          req_acks=SimpleProducer.ACK_AFTER_LOCAL_WRITE,
+                          ack_timeout=2000)
+
+
+conf = urlparse.urlparse(os.environ['RDS_URL'])
+mysql_settings = {'host': conf.hostname, 
+					'port': conf.port, 
+					'user': conf.username, 
+					'passwd': conf.password}
+
+# Connect to Mysql replication stream
+print "Connecting to Mysql at {}...".format(mysql_settings['host'])
+stream = BinLogStreamReader(connection_settings = mysql_settings, server_id=100, resume_stream=False,
+                                blocking=True)
+print "connected. Listening for changes..."
+
+for evt in stream:
+	evt.dump()
+	msg = build_message(evt)
+	if msg:
+		try:
+			response = producer.send_messages(msg['headers']['table'], json.dumps(msg, default=json_serial))
+		except LeaderNotAvailableError:
+			time.sleep(1)
+			response = producer.send_messages(msg['headers']['table'], json.dumps(msg, default=json_serial))
+		# TODO: Test response.error
+		# TODO: Store replication stream pos
+
+stream.close()

package.json

@@ -0,0 +1,20 @@
+{
+  "name": "kafka-data-listener",
+  "version": "1.0.0",
+  "description": "Simple Mysql replication stream consumer which publishes change events to Kafka.",
+  "main": "consumer.js",
+  "scripts": {
+    "test": "echo \"Error: no test specified\" && exit 1"
+  },
+  "author": "",
+  "license": "ISC",
+  "dependencies": {
+    "express": "^4.12.3",
+    "glob": "^5.0.5",
+    "kafka-node": "^0.2.26",
+    "knex": "^0.8.3",
+    "pg": "^4.3.0",
+    "pg-hstore": "^2.3.2",
+    "socket.io": "^1.3.5"
+  }
+}

requirements.txt

@@ -0,0 +1,2 @@
+kafka-python==0.9.3
+mysql-replication==0.5

scripts/magento.core_session.js

@@ -0,0 +1,12 @@
+module.exports = {
+	'magento.core_session': function(context, msg) {
+		console.log("Session hit: id: " + msg.data.session_id + ", expires: " + new Date(msg.data.session_expires * 1000));
+	},
+	'magento.sales_flat_quote_item': function(context, msg) {
+		console.log("Added to cart: " + msg.data.name);
+	},
+	'magento.log_url': function(context, msg) {
+		console.log("Page hit on url " + msg.data.url_id + " by visitor " + msg.data.visitor_id + " at " +
+			msg.data.visit_time);
+	}
+}