JDBC Source Connector

Quickstart

数据库环境准备

CREATE DATABASE connector;
USE connector;
CREATE TABLE `from_source` (
  `fdsid` int(11) NOT NULL AUTO_INCREMENT,
  `dsid` int(11) DEFAULT NULL,
  `from` int(11) DEFAULT NULL,
  `stype` int(11) DEFAULT NULL,
  PRIMARY KEY (`fdsid`)
) ENGINE=InnoDB AUTO_INCREMENT=6 DEFAULT CHARSET=utf8mb4;
INSERT INTO from_source VALUES(1, 2, 1, 1);
INSERT INTO from_source VALUES(2, 2, 1, 2);
INSERT INTO from_source VALUES(3, 2, 1, 5);
INSERT INTO from_source VALUES(4, 2, 1, 6);

MySQL JDBC 驱动准备

测试环境使用的mysql版本信息如下：

mysql-info.png

在mysql官网上选择合适的驱动下载，测试中下载的是mysql-connector-java-5.1.42.tar.gz。
将此驱动拷贝到$CONFLUENT_PATH/share/java/kafka-connect-jdbc目录下，使用解压命令解压

tar -xzvf mysql-connector-java-5.1.42.tar.gz
cd mysql-connector-java-5.1.42
cp mysql-connector-java-5.1.42-bin.jar ../

最终的目的就是将mysql-connector-java-5.1.42-bin.jar放在$CONFLUENT_PATH/share/java/kafka-connect-jdbc目录下，这样confluecnt connector在启动是就可以找到mysql的jdbc驱动了

配置文件准备

单机环境下运行connector的命令如下：

//INFO Usage: ConnectStandalone worker.properties connector1.properties [connector2.properties ...] (org.apache.kafka.connect.cli.ConnectStandalone:61)
bin/connect-standalone etc/schema-registry/connect-avro-standalone.properties etc/kafka-connect-jdbc/mysql-source.properties

其中:connect-avro-standalone.properties可使用默认配置；
mysql-source.properties的内容如下：

# tasks to create:
name=test-mysql-jdbc-autoincrement
connector.class=io.confluent.connect.jdbc.JdbcSourceConnector
tasks.max=1
# a table called 'users' will be written to the topic 'test-mysql-jdbc-users'.
connection.url=jdbc:mysql://172.24.8.114:3306/connector?user=$USER&password=$PASSWORD
mode=incrementing
incrementing.column.name=fdsid
topic.prefix=test-mysql-jdbc-

其中，$USER,$PASSWORD是访问mysql数据库的用户名和地址。
参考：
JDBC驱动下载地址
 MySQL Connector配置参考

从MySQL导入数据到Kafka中

启动Connector

 bin/connect-standalone etc/schema-registry/connect-avro-standalone.properties etc/kafka-connect-jdbc/mysql-source.properties

验证产生了相应的topic

bin/kafka-topics --zookeeper localhost:2181 --list

topic列表中会包含：test-mysql-jdbc-from_source

验证topic中的数据

bin/kafka-avro-console-consumer --new-consumer --bootstrap-server localhost:9092 --topic test-mysql-jdbc-from_source --from-beginning

此时，可以获取topic:test-mysql-jdbc-from_source中的所有数据。

验证可增量从MySQL中导入数据

向from_source表中添加一个数据

insert into from_source values(7, 2, 1, 1)

相应的消费者会接收到插入到from_source表中的数据。
总参考：
JDBC Source Connector 官网

confluent jdbc connector - mysql