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放在一个文件夹下，打开cmd命令行，输入docker build -t curtain:kafka-connector:7.0.1 .构建连接器镜像

version: "3.0"
services:
  zookeeper:
    image: confluentinc/cp-zookeeper:7.0.1
    container_name: zookeeper
    restart: always
    ports:
      - 2181:2181
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000
      ZOOKEEPER_SYNC_LIMIT: 2
  kafka:
    image: confluentinc/cp-kafka:7.0.1
    container_name: kafka
    ports:
      - 9092:9092
    environment:
      # client要访问的broker地址
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://服务器ip:9092
      KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9092
      # 通过端口连接 zookeeper
    restart: always
    depends_on:
      - zookeeper
  kafka-connector:
    image: curtain/kafka-connector:7.0.1
    container_name: kafka-connector
    volumes:
      - type: bind
        source: ./connect-standalone.properties
        target: /etc/kafka/connect-standalone.properties
      - type: bind
        source: ./connect-redis-source.properties
        target: /etc/kafka/connect-redis-source.properties
    restart: always
    depends_on:
      - zookeeper
      - kafka

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

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

1. 加载镜像：docker load < 镜像包.tar
2. 启动镜像： 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即可。