kafka connect简述以及简单实用以及mysql导入ES

左手的ㄟ右手 2022-04-05 09:40 226阅读 0赞

#### 简述 ####

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

#### 启动connect ####

*  创建配置文件  
    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自带的插件很少，需要后面我们自己新加插件。

*  添加插件
    
     *  去官网下载ES和JDBC的插件，官网地址：[https://www.confluent.io/connector/kafka-connect-jdbc/][https_www.confluent.io_connector_kafka-connect-jdbc]
     *  下载下来应该是两个ZIP包，解压到插件目录
     *  JDBC包默认只支持sqlite和PG,所以需要添加mysql-connector-java-x.x.x.jar驱动到 confluentinc-kafka-connect-jdbc-5.0.1/lib 目录下
 *  启动connect

./connect-distributed.sh -daemon /app/kafka/config/connect-distributed.properties

*  验证是否启动成功

curl -X GET -i 'http://x.x.x.x:8083/connector-plugins'

**正常返回**

[{"class":"io.confluent.connect.elasticsearch.ElasticsearchSinkConnector","type":"sink","version":"5.0.1"},{"class":"io.confluent.connect.jdbc.JdbcSinkConnector","type":"sink","version":"5.0.1"},{"class":"io.confluent.connect.jdbc.JdbcSourceConnector","type":"source","version":"5.0.1"},{"class":"org.apache.kafka.connect.file.FileStreamSinkConnector","type":"sink","version":"2.1.0"},{"class":"org.apache.kafka.connect.file.FileStreamSourceConnector","type":"source","version":"2.1.0"}]

#### Mysql数据导入Kafka ####

*  mysql表概览  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTQ2ODYzOTk_size_16_color_FFFFFF_t_70]
 *  启动连接器

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这个主题，如果能看到输出说明你的连接器配置成功了。

#### Kafka导入ES ####

*  启动连接器

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的值会自动生成。

*  查看是否获取到数据  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTQ2ODYzOTk_size_16_color_FFFFFF_t_70 1]

#### 增量导入 ####

*  向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  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTQ2ODYzOTk_size_16_color_FFFFFF_t_70 2]

--------------------

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

[https_www.confluent.io_connector_kafka-connect-jdbc]: https://www.confluent.io/connector/kafka-connect-jdbc/
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTQ2ODYzOTk_size_16_color_FFFFFF_t_70]: /images/20220405/ec29fca1991b4b25bbb3fe5c2f13d396.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTQ2ODYzOTk_size_16_color_FFFFFF_t_70 1]: /images/20220405/c751d5ca2c7f473088dd0a613ae7fa9b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTQ2ODYzOTk_size_16_color_FFFFFF_t_70 2]: /images/20220405/672c9662acd44e9190119d5384ae7b66.png