2014 年的時候，Kafka 的三個主要開發人員從 LinkedIn 出來創業，開了一家叫作 Confluent 的公司。和其他大資料公司類似，Confluent 的產品叫作 Confluent Platform。這個產品的核心是 Kafka，分為三個版本：Confluent Open Source、Confluent Enterprise 和 Confluent Cloud。

安裝jdbc-mysql-driver

wget http://dev.mysql.com/get/Downloads/Connector-J/mysql-connector-java-5.1.39.tar.gz
tar xzvf mysql-connector-java-5
 
.1.39.tar.gz
sed -i '$a export CLASSPATH=/root/mysql-connector-java-5.1.39/mysql-connector-java-5.1.39-bin.jar:$CLASSPATH' /etc/profile
source /etc/profile

安裝confluent

下載confluent的tar包解壓安裝。

cd /usr/local
# tar zxvf confluent.tar.gz

confluent平臺各元件的預設埠號

confluent的mysql資料來源配置

建立一個confluent從mysql載入資料的配置檔案quickstart-mysql.properties

name=mysql-whitelist-timestamp-source
connector.class=io.confluent.connect.jdbc.JdbcSourceConnector
tasks.max=10
connection.user=root
connection.password=root
connection.url=jdbc:
 
mysql://192.168.248.128:3306/foodsafe?characterEncoding=utf8&useSSL=true

#資料表白名單
#table.whitelist=t1

mode=timestamp+incrementing
timestamp.column.name=modified
incrementing.column.name=id

#topic的字首，confulent平臺會為每張表建立一個topic,topic的名稱為字首+表名
topic.prefix=mysql-test-

自定義查詢模式：

如果使用上面的配置來啟動服務，則confluent平臺將會監測拉取所有表的資料，有時候可能並不需要這樣做，confulent平臺提供了自定義查詢模式。配置參考如下：

#User defined connector instance name
name=mysql-whitelist-timestamp-source
#The classimplementingtheconnector
connector.class=io.confluent.connect.jdbc.JdbcSourceConnector
#Maximum number of tasks to run for this connector instance
tasks.max=10

connection.url=jdbc:mysql://192.168.248.128:3306/foodsafe?characterEncoding=utf8&useSSL=true
connection.user=root
connection.password=root
query=SELECT f.`name`,p.price,f.create_time from foods f join price p on (f.id = p.food_id)
mode=timestamp
timestamp.column.name=timestamp

topic.prefix=mysql-joined-data

query模式下使用where查詢語句容易造成kafka拼接sql錯誤，最好採用join

1.啟動zookeeper

因為zookeeper是一個長期的服務，最好在後臺執行，同時需要有寫許可權到/var/lib在這一步以及之後的步驟，如果沒有許可權請檢視安裝confulent的使用者是否具有/var/lib的寫許可權

# cd /usr/local/confulent-3.2.2
# ./bin/zookeeper-server-start ./etc/kafka/zookeeper.properties &
# 以守護程序方式啟動
# sudo confluent-3.2.2/bin/zookeeper-server-start -daemon /etc/kafka/zookeeper.properties

停止zookeeper

$ ./bin/zookeeper-server-stop

2.啟動kafka

# cd /usr/local/confluent-3.2.2
# ./bin/kafka-server-start ./etc/kafka/server.properties &

停止kafka服務

./bin/kafka-server-stop

3.啟動Schema Registry

# cd /usr/local/confluent-3.2.2
# ./bin/schema-registry-start ./etc/schema-registry/schema-registry.properties &

停止schema-registry

# ./bin/schema-registry-stop

4.啟動監聽mysql資料的producer

# cd /usr/local/confluent-3.2.2
# ./bin/connect-standalone ./etc/schema-registry/connect-avro-standalone.properties ./etc/kafka-connect-jdbc/quickstart-mysql.properties &

5.啟動消費資料的consumer

# cd /usr/local/confluent-3.2.2
#./bin/kafka-avro-console-consumer --new-consumer --bootstrap-server localhost:9092 --topic mysql-test-t1 --from-beginning

測試sql

DROP TABLE IF EXISTS `t1`;
CREATE TABLE `t1` (
  `id` int(11) NOT NULL AUTO_INCREMENT,
  `name` varchar(200) DEFAULT NULL,
  `createtime` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
  `modified` timestamp NOT NULL DEFAULT '0000-00-00 00:00:00' ON UPDATE CURRENT_TIMESTAMP,
  PRIMARY KEY (`id`),
  KEY `id` (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=9 DEFAULT CHARSET=utf8;

-- ----------------------------
-- Records of t1
-- ----------------------------
INSERT INTO `t1` VALUES ('1', 'aa', '2017-07-10 08:03:51', '2017-07-10 23:03:30');
INSERT INTO `t1` VALUES ('3', 'bb', '2017-07-10 08:03:45', '2017-07-10 23:03:34');
INSERT INTO `t1` VALUES ('4', '年內', '2017-07-10 08:05:51', '2017-07-10 23:05:45');
INSERT INTO `t1` VALUES ('5', '年內', '2017-07-10 08:44:28', '2017-07-10 23:15:45');
INSERT INTO `t1` VALUES ('6', '公共', '2017-07-18 06:05:11', '2017-07-18 21:04:58');
INSERT INTO `t1` VALUES ('7', '哈哈', '2017-07-18 19:05:04', '2017-07-18 07:32:13');
INSERT INTO `t1` VALUES ('8', '公共經濟', '2017-07-27 20:33:10', '2017-07-18 07:34:43');

資料插入語句

INSERT INTO `t1` (name,createtime,modified)VALUES ('公共經濟2', '2017-07-27 20:33:10', '2017-07-18 07:34:43');

插入新資料後將會在consumer端實時輸出我們插入的資料

{"id":7,"name":{"string":"哈哈"},"createtime":1500429904000,"modified":1500388333000}
{"id":8,"name":{"string":"公共經濟"},"createtime":1501212790000,"modified":1500388483000}
{"id":9,"name":{"string":"公共經濟1"},"createtime":1501212790000,"modified":1500388483000}
{"id":10,"name":{"string":"公共經濟2"},"createtime":1501212790000,"modified":1500388483000}

關於confluent的使用國內目前使用似乎很少，相關的中文文件也極少。本文是去年7月份我在做實時資料交換技術調研是根據官方文件實踐的記錄。