kafka connector使用（Docker一键启动版）

前言

结合博主之前写的一篇《kafka connector使用（单机手动启动版）》一起看

版本

kafka: confluentinc/cp-kafka:7.0.1
zookeeper: confluentinc/cp-zookeeper:7.0.1

思路

将该连接器做成一个镜像。

实现

一、制作镜像

Dockerfile

FROM confluentinc/cp-kafka:7.0.1
COPY connect-redis.jar /usr/share/java/connect-redis.jar
CMD ["connect-standalone","/etc/kafka/connect-standalone.properties","/etc/kafka/connect-redis-source.properties"]

将Dockerfile和超级jar放在一个文件夹下，docker build -t curtain/kafka-connector:7.0.1 .构建连接器镜像

二、编写docker-compose.yml

version: "3.0"
services:zookeeper:image: confluentinc/cp-zookeeper:7.0.1container_name: zookeeperrestart: alwaysports:- 2181:2181environment:ZOOKEEPER_CLIENT_PORT: 2181ZOOKEEPER_TICK_TIME: 2000ZOOKEEPER_SYNC_LIMIT: 2kafka:image: confluentinc/cp-kafka:7.0.1container_name: kafkaports:- 9092:9092environment:# client要访问的broker地址KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://服务器ip:9092KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9092# 通过端口连接 zookeeperrestart: alwaysdepends_on:- zookeeperkafka-connector:image: curtain/kafka-connector:7.0.1container_name: kafka-connectorvolumes:- type: bindsource: ./connect-standalone.propertiestarget: /etc/kafka/connect-standalone.properties- type: bindsource: ./connect-redis-source.propertiestarget: /etc/kafka/connect-redis-source.propertiesrestart: alwaysdepends_on:- zookeeper- kafka

三、修改connect-standalone.properties

没有该文件的话，网上找一份。将bootstrap.servers里面的ip部分从localhost改成服务器的ip。

四、上传文件

将镜像包、docker-compose.yml、connect-standalone.properties、连接器配置文件上传到服务器同一个路径下。

五、加载镜像并启动

加载镜像：docker load < 镜像包.tar
启动镜像： docker-compose up -d

这样，kafka连接器就启动啦。

可能会遇到的问题

问题一：默认写入到kafka的数据会包含一些额外的信息如：schema和payload，我们不想要这些额外的信息怎么办？
将connect-standalone.properties配置文件中的key.converter.schemas.enable和value.converter.schemas.enable设置为false即可
问题二：写入到kafka的数据多了一层双引号和json转义怎么去掉？
这个是因为默认配置的转换器类是org.apache.kafka.connect.json.JsonConverter加上你生成的SourceRecord对象里面的value和key值已经是一个String类型了。这样的话会将String类型的数据json化一遍，就会出现多了一层双引号加转义的情况。修改connect-standalone.properties配置文件中的key.converter和value.converter为org.apache.kafka.connect.storage.StringConverter即可。