kafka connnect知识解析和相关连接操作

小灰灰 2022-11-28 15:10 77阅读 0赞

## 一、基础知识 ##

### 1.Connectors 和Tasks ###

首先Connectors分为两类：SourceConnectors 和  SinkConnectors。SourceConnectors就是从其他系统导入数据到Kafka上，而SinkConnectors就是将Kafka上的数据导出到其他系统。例如，JDBCSourceConnector就是将关系型数据库中的数据导入到Kafka上，HDFSSinkConnector就是将Kafka上的数据导出到HDFS上。

Connectors类的实现本身不执行数据复制：Connector class的配置描述了要复制的数据集，Connector负责将该作业分解成一组Tasks，这些Tasks可以分发给Kafka Connect workers。

其次Tasks也分为两类：SourceTask 和 SinkTask。对于要分配的数据每一个Task都会复制数据的子集到Kafka（或从kafka复制数据）。connector复制的数据表示为一个分区流，类似于Kafka topic的模型，其中每一个分区都是带有偏移量的有序记录序列。每个task被分配到要处理的分区的一个子集。

下图是source connector的实例，它创建了两个tasks,tasks从输入分区（Stream中）拷贝数据并且将数据写到Kafka中

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjA3MDQ3Mw_size_16_color_FFFFFF_t_70][]

下图是数据通过连接source task进入Kafka的高级表示.

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjA3MDQ3Mw_size_16_color_FFFFFF_t_70 1][]

### 2.Partitions and Records ###

每个分区都是key-value记录的有序序列。keys和values可以有复杂的结构，由org.apache.kafka.connect中的数据结构表示包数据。支持许多基本类型以及数组、结构和嵌套数据结构。

下图是一个分区流，这是分区流数据模型用connectors对所有source和sink进行映射。每个记录包含keys和values(带有schemas)、分区ID和该分区内的偏移量（offsets）。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjA3MDQ3Mw_size_16_color_FFFFFF_t_70 2][]

## 二、具体操作 ##

### 1.配置 ###

用Kafka connect连接其他系统时，需要根据传输的数据指定序列化数据的格式。在<Confluent-Path>etc/schema-registry/connect-distributed.properties 配置文件中修改。 现有的数据格式有JSON、AVRO、Protobuf等。

若是AVRO:（注：此处用的是Confluent5.5.1）

key.converter=io.confluent.connect.avro.AvroConverter
    key.converter.schema.registry.url=http://localhost:8081
    value.converter=io.confluent.connect.avro.AvroConverter
    value.converter.schema.registry.url=http://localhost:8081

若是JSON:

key.converter=io.confluent.connect.json.JsonSchemaConverter
    key.converter.schema.registry.url=http://localhost:8081
    value.converter=io.confluent.connect.json.JsonSchemaConverter
    value.converter.schema.registry.url=http://localhost:8081

### 2.启动 ###

1）kafka connect（distributed模式）启动命令：

①bin/connect-distributed.sh -daemon config/connect-distributed.properties

②bin/connect-distributed.sh  config/connect-distributed.properties

启动后验证网址：[http://主机IP:8083/connector-plugins][http_IP_8083_connector-plugins]

2)通过curl命令连接（可以是别的方式我选择的是curl）

以连接mqtt为例，

curl -s -X POST -H 'Content-Type: application/json' http://主机IP:8083/connectors -d '{
        "name" : "连接名",
    "config" : {
        "connector.class" : "io.confluent.connect.mqtt.MqttSourceConnector",
        "tasks.max" : "1",
        "mqtt.server.uri" : "tcp://mqtt服务器IP:1883",
        "mqtt.topics" : "mqtt的主题",
        "kafka.topic" : "kafka的主题",
        "confluent.topic.bootstrap.servers": "主机ip:9092",
        "confluent.topic.replication.factor": "3",
        "value.converter" :
         "org.apache.kafka.connect.converters.ByteArrayConverter"
        }
    }'

3）查看连接其运行状态

[http://主机IP:8083/connector-plugins][http_IP_8083_connector-plugins]/连接器名/status

### 3.其余操作 ###

若采用JSON格式传输数据的话，kafka consumer会收到如下的数据：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjA3MDQ3Mw_size_16_color_FFFFFF_t_70 3][]![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjA3MDQ3Mw_size_16_color_FFFFFF_t_70 4][]

其中只有key为“payload”对应的value是我们传输的真实数据，所以我们可以利用Kafka consumer的API操作过滤的其余数据，只获取“payload”对应的数据。

while (true) {
    
                //拉取数据
                ConsumerRecords<String, String> consumerRecords = 
                              consumer.poll(Duration.ofMillis(100));
                //解析并打印consumerRecords
                for (ConsumerRecord<String,String> record : consumerRecords) {
                    JSONObject jo= JSON.parseObject(record.value()); //解析JSON对象
                    String val=jo.getString("payload");   //获取payload对应的value
                    System.out.println(val);
    
                    ProducerRecord<String,String> message=new ProducerRecord<>("test1",
                            val);       //将获取的value发送到新的kafka topic 
                    producer.send(message);
    
    
                }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjA3MDQ3Mw_size_16_color_FFFFFF_t_70]: /images/20221124/b574fc1cc4a24412b76e3d618e4afccc.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjA3MDQ3Mw_size_16_color_FFFFFF_t_70 1]: /images/20221124/f547c9cb1ed6438d9ae11c85360eefdb.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjA3MDQ3Mw_size_16_color_FFFFFF_t_70 2]: /images/20221124/3342d00e209940fa89cdc0cf252d91d9.png
[http_IP_8083_connector-plugins]: http://192.168.133.101:8083/connector-plugins
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjA3MDQ3Mw_size_16_color_FFFFFF_t_70 3]: https://img-blog.csdnimg.cn/20201224111140389.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjA3MDQ3Mw==,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjA3MDQ3Mw_size_16_color_FFFFFF_t_70 4]: /images/20221124/6be94428a9b54330a5ac564348a06699.png