• 用途
  connect是kafka的一部分，用于和外部存储进行交换数据（可以导入也可以导出），类似于ES的logstash。估计你会有疑问，我可以写程序定制消费者或者生产者完成这样的功能呀，为啥用这个呀？答案很简单：就是方便，它帮你完成了很多细节比如连接数据库和关闭连接等等。
• 组成
  • 连接器
    用于配置和分配任务。比如JDBC连接器会连接到数据库，统计需要操作的数据表的数据来决定启动多少个任务，然后为每个任务生成配置（比如：数据的连接地址或者文件的路径）
  • 任务
    任务是数据的搬运工。负责将数据移入和移除kafka
  • worker进程
    是连接器和任务的容器。负责接收HTTP请求，并处理请求，定义连接器的配置并启动连接器

• 创建配置文件
  connect的配置文件在config目录下，connect-distributed.properties connect-standalone.properties,前者是多节点的配置文件，后者是单节点的配置文件。本文用前者。
• 编辑connect-distributed.properties

bootstrap.servers=xxxx:9092,xxxx:9092,xxxx:9092
group.id=connect-han
key.converter=org.apache.kafka.connect.json.JsonConverter
value.converter=org.apache.kafka.connect.json.JsonConverter
key.converter.schemas.enable=true
value.converter.schemas.enable=true
offset.storage.topic=connect-offsets
offset.storage.replication.factor=1
config.storage.topic=connect-configs
config.storage.replication.factor=1
status.storage.topic=connect-status
status.storage.replication.factor=1
offset.flush.interval.ms=10000
rest.port=8083
plugin.path=/app/kafka/connect-plugin

bootstrap.servers 这里是broker的地址；group.id这是connect集群的名字，集群内的多个节点应该一样；rest.port这是connect监听的端口，默认就是8083；plugin.path 这是插件的目录，因为connect自带的插件很少，需要后面我们自己新加插件。

• mysql表概览
  
• 启动连接器

curl -X POST -H 'Content-Type: application/json' -i 'http://x.x.x.x:8083/connectors' --data '{"name":"load-mysql-data",
"config":{
"connector.class":"io.confluent.connect.jdbc.JdbcSourceConnector",
"connection.url":"jdbc:mysql://x.x.x.x:3306/esdatabase?user=esuser&password=123.com",
"table.whitelist":"product",
"incrementing.column.name": "product_id",
"mode":"incrementing",
"topic.prefix": "test-mysql-jdbc-"
}
}'

name每个连接器的名字是唯一的；connector.class这里用你的JDBC插件的名字；connection.url jdbc的配置；table.whitelist 要导入的表的名字，可以写多个用逗号分隔；incrementing.column.name 自增列的名字，用于增量导入；topic.prefix 创建topic的前缀，这里创建的topic就是 test-mysql-jdbc-proudct 。

• 验证
  你可以启动一个消费者，从起始点开始消费test-mysql-jdbc-proudct这个主题，如果能看到输出说明你的连接器配置成功了。

• 启动连接器

curl -X POST -H 'Content-Type: application/json' -i 'http://x.x.x.x:8083/connectors' --data '{"name":"dump-es",
"config":{
"connector.class":"io.confluent.connect.elasticsearch.ElasticsearchSinkConnector",
"connection.url":"http://x.x.x.x:9200",
"type.name":"doc",
"key.ignore": "true",
"topics":"test-mysql-jdbc-product"
}
}'

如果你的ES不允许自动创建索引，你需要提前创建一个索引，名字就是topics的值，字段的名字要对应好mysql表。
connector.class这里要写你的ES插件的地址；connection.url这里写ES的地址；type.name类型的名字；key.ignore如果设置为true，ES里面_id的值会自动生成。

• 查看是否获取到数据
  

• 向mysql插入数据

insert product(product_name,product_price, product_color,brand_name) value ("hat",76,"red",1);
insert product(product_name,product_price, product_color,brand_name) value ("hat",126,"green",2);

• 查看ES
  

先写到这里了，有问题进QQ群630300475