Confluent介绍及其使用

1 confluent介绍

Confluent是用来管理和组织不同数据源的流媒体平台，可以实时地把不同源和位置的数据集成到一个中心的事件流平台。并且很可靠、性能很高。

Confluent目前提供了社区版（免费）和商业版（收费）两个版本，社区版提供了Connectors、REST Proxy、KSQL、Schema-Registry等基础服务。商业版为企业提供了控制面板、负载均衡，跨中心数据备份、安全防护等高级特性。

1.2 服务功能介绍

1.2.1 Zookeeper

Zookeeper是一个开放源码的分布式应用程序协调服务，主要功能包扩：维护配置信息、命名、提供分布式同步、组管理等集中式服务。Kafka使用ZooKeeper对集群元数据进行持久化存储，如果ZooKeeper丢失了Kafka数据，集群的副本映射关系以及topic等配置信息都会丢失，最终导致Kafka集群不再正常工作，造成数据丢失的后果。

1.2.2 Kafka

Kafka是一个分布式流处理平台，基于zookeeper协调并支持分区和多副本的分布式消息系统，是一种高吞吐量的分布式发布订阅消息系统，消息队列中间件，主要功能是负责消息传输，Confluent就是依赖Kafka来进行消息传输。Kafka最大的特性就是可以实时的处理大量数据以满足各种需求场景。

1.2.3 Control Center

control center可以很容易地管理kafka的连接，创建，编辑，和管理与其他系统的连接。我们可以从producer到consumer监控data streams，保证我们的每一条消息都被传递，还能测量出消息的传输耗时多久。使用confluent control center能让开发人员不写一句代码，也能构建基于kafka的数据生产管道。

1.2.4 Kafka-rest

Kafka-rest是Kafka RESTful接口服务组件，可以通过Restful接口而不是本机Kafka协议或客户端的情况下，生成和使用消息,而且还可以查看集群状态以及执行管理操作。

1.2.5 Schema-Registry

Schema-Registry是为元数据管理提供的服务，同样提供了RESTful接口用来存储和获取schemas，它能够保存数据格式变化的所有版本，并可以做到向下兼容。Schema-Registry还为Kafka提供了Avro格式的序列化插件来传输消息。Confluent主要用Schema-Registry来对数据schema进行管理和序列化操作。

1.2.6 Connect

Kafka Connect是 Kafka的一个开源组件，是用来将Kafka与数据库、key-value存储系统、搜索系统、文件系统等外部系统连接起来的基础框架。通过使用Kafka Connect框架以及现有的连接器可以实现从源数据读入消息到Kafka，再从Kafka读出消息到目的地的功能。

1.2.7 ksql-server

KSQL是使用SQL语句对Apache Kafka执行流处理任务的流式SQL引擎，Confluent 使用KSQL对Kafka的数据提供查询服务.

2 confluent下载

使用的开源的confluent的5.2.4版本

下载链接：http://packages.confluent.io/archive/5.2/confluent-5.2.4-2.11.tar.gz

3 环境准备

分布式搭建建议至少3个节点，但是由于用于测试及节点紧张这里使用2个节点

节点	zookeeper	kafka	control-center	kafka-reset	schema-registry	connector	ksql-server
10.0.165.8	√	√	√	√	√	√	√
10.0.165.9	√	√	√	√	√	√	√
	2181	9092	9021	8082	8081	8083	8088

4 安装

4.1 解压

将下载的文件上传至linux，然后解压至相应的目录下

tar -zxvf /opt/package/confluent-5.2.4-2.11.tar.gz -C /home/kafka/.local/

修改文件名并进入到相应的目录下

mv /home/kafka/.local/confluent-5.2.4 /home/kafka/.local/confluent
cd /home/kafka/.local/confluent

4.2 修改配置

修改10.0.165.8节点的相应配置

4.2.1 zookeeper配置

(1)vim /home/kafka/.local/confluent/etc/kafka/zookeeper.properties

##数据存放目录，默认为/tmp/zookeepe存在删除风险
dataDir=/data/confluent/zookeeper
clientPort=2181
maxClientCnxns=0
initLimit=5
syncLimit=2##多个zookeeper server，server的编号1、2等要与myid中的一致
server.1=10.0.165.8:2888:3888
server.2=10.0.165.9:2888:3888

(2)生成myid

echo 1 > /home/kafka/.local/confluent/etc/kafka/myid

(3)修改confluent服务启动脚本，将myid发布到confluent运行目录下。

bin/confluent start会启动confluent的各服务，且会将etc下的各配置，复制到confluent运行目录下。

vim /home/kafka/.local/confluent/bin/confluent

在config_zookeeper()方法块最后一行，添加

cp ${confluent_conf}/kafka/myid $confluent_current/zookeeper/data/

目的是将etc/kafka/myid拷贝到confluent运行目录下，否则会报myid is no found，zookeeper启动失败。

4.2.2 Kafka配置

vim /home/kafka/.local/confluent/etc/kafka/server.properties

broker.id=0#listeners与advertised.listeners可以只配一个，与当前机器网卡有关系，请注意。advertised.listeners可能通用性更强，值为当前机器的ip与端口，其他机器ip无需配置
advertised.listeners=PLAINTEXT://10.0.165.8:9092##根据实际情况调整
num.network.threads=8
num.io.threads=8
socket.send.buffer.bytes=1048576
socket.receive.buffer.bytes=1048576
socket.request.max.bytes=104857600
fetch.purgatory.purge.interval.requests=100
producer.purgatory.purge.interval.requests=100#log.dirs是最重要的配置，kafka数据所在
log.dirs=/data/confluent/kafka-logs
num.partitions=12num.recovery.threads.per.data.dir=1message.max.bytes=10000000
replica.fetch.max.bytes= 10485760
auto.create.topics.enable=true
auto.leader.rebalance.enable = true##备份因子数<=kafka节点数，若大于会报错
default.replication.factor=2
offsets.topic.replication.factor=2
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1log.flush.interval.messages=20000
log.flush.interval.ms=10000
log.flush.scheduler.interval.ms=2000
log.retention.check.interval.ms=300000
log.cleaner.enable=true##log失效时间，单位小时
log.retention.hours=48
zookeeper.connect=10.0.165.8:2181,10.0.165.9:2181
zookeeper.connection.timeout.ms=6000
zookeeper.sync.time.ms=2000confluent.metrics.reporter.bootstrap.servers=10.0.165.8:9092,10.0.165.9:9092
confluent.metrics.reporter.topic.replicas=2confluent.support.metrics.enable=true
confluent.support.customer.id=anonymousdelete.topic.enable=true
group.initial.rebalance.delay.ms=0

4.2.3 kafka-rest

vim /home/kafka/.local/confluent/etc/kafka-rest/kafka-rest.properties

id=kafka-rest-server-001
schema.registry.url=http://10.0.165.8:8081
zookeeper.connect=10.0.165.8:2181,10.0.165.9:2181
bootstrap.servers=PLAINTEXT://10.0.165.8:9092
port=8082
consumer.threads=8access.control.allow.methods=GET,POST,PUT,DELETE,OPTIONS
access.control.allow.origin=*

4.2.4 ksql

confluent-4没有这个

vim /home/kafka/.local/confluent/etc/ksql/ksql-server.properties

ksql.service.id=default_
bootstrap.servers=10.0.165.8:9092,10.0.165.9:9092
listeners=http://0.0.0.0:8088
ksql.schema.registry.url=http://10.0.165.8:8081,http://10.0.165.9:8081
ksql.sink.partitions=4

4.2.5 confluent-control-center

vim /home/kafka/.local/confluent/etc/confluent-control-center/control-center-dev.properties

bootstrap.servers=10.0.165.8:9092,10.0.165.9:9092
zookeeper.connect=10.0.165.8:2181,10.0.165.9:2181
confluent.controlcenter.rest.listeners=http://0.0.0.0:9021#每个id要唯一，不然只能启动一个
confluent.controlcenter.id=1
confluent.controlcenter.data.dir=/data/confluent/control-center
confluent.controlcenter.connect.cluster=http://10.0.165.8:8083,http://10.0.165.9:8083##每台都配置各自的ip
confluent.controlcenter.ksql.url=http://10.0.165.8:8088
confluent.controlcenter.schema.registry.url=http:/10.0.165.8:8081,http://10.0.165.9:8081confluent.controlcenter.internal.topics.replication=2
confluent.controlcenter.internal.topics.partitions=2
confluent.controlcenter.command.topic.replication=2
confluent.monitoring.interceptor.topic.partitions=2
confluent.monitoring.interceptor.topic.replication=2
confluent.metrics.topic.replication=2confluent.controlcenter.streams.num.stream.threads=30

4.2.6 schema-registry

vim /home/kafka/.local/confluent/etc/schema-registry/schema-registry.properties

listeners=http://0.0.0.0:8081
kafkastore.bootstrap.servers=PLAINTEXT://10.0.165.8:9092,10.0.165.9:9092
kafkastore.topic=_schemas
debug=false

4.2.7 connect

vim /home/kafka/.local/confluent/etc/schema-registry/connect-avro-distributed.properties

bootstrap.servers=10.0.165.8:9092,10.0.165.9:9092
group.id=connect-clusterkey.converter=org.apache.kafka.connect.storage.StringConverter
key.converter.schema.registry.url=http://localhost:8081
value.converter=io.confluent.connect.avro.AvroConverter
value.converter.schema.registry.url=http://localhost:8081config.storage.topic=connect-configs
offset.storage.topic=connect-offsets
status.storage.topic=connect-statusesconfig.storage.replication.factor=2
offset.storage.replication.factor=2
status.storage.replication.factor=2internal.key.converter=org.apache.kafka.connect.json.JsonConverter
internal.value.converter=org.apache.kafka.connect.json.JsonConverter
internal.key.converter.schemas.enable=false
internal.value.converter.schemas.enable=falserest.port=8083
rest.advertised.port=8083plugin.path=/home/kafka/.local/confluent/share/java

4.2.8 将confluent发送到其他节点

scp -r confluent/ kafka@10.0.165.9:/home/kafka/.local/

然后修改其他节点的配置

vi myid

vi /home/kafka/.local/confluent/etc/kafka/server.properties

broker.id=1
advertised.listeners=PLAINTEXT://10.0.165.9:9092

vi /home/kafka/.local/confluent/etc/kafka-rest/kafka-rest.properties

id=kafka-rest-server-002
schema.registry.url=http://10.0.165.9:8081
bootstrap.servers=PLAINTEXT://10.0.165.9:9092

vi /home/kafka/.local/confluent/etc/confluent-control-center/control-center-dev.properties

confluent.controlcenter.id=2
confluent.controlcenter.ksql.url=http://10.0.165.9:8088

然后在两个节点的/data目录下新建confluent并修改权限

sudo mkdir /data/confluent
sudo chown kafka:kafka /data/confluent

4.3 服务启动与停止

4.3.1 全部服务启动

启动：bin/confluent start

查看状态：bin/confluent status

停止：bin/confluent stop

4.3.2 单独启动服务

服务单独启动

启动kafka-rest

bin/kafka-rest-start   etc/kafka-rest/kafka-rest.properties

上面的这种方式是前台启动，也可以以后台方式启动。

nohup bin/kafka-rest-start   etc/kafka-rest/kafka-rest.properties &

启动zookeeper

bin/zookeeper-server-start -daemon etc/kafka/zookeeper.properties

启动kafka broker

bin/kafka-server-start -daemon  etc/kafka/server.properties

启动schema registry

bin/schema-registry-start -daemon  etc/schema-registry/schema-registry.properties

5 安装过程常见报错

5.1 KafkaServer启动失败

[2020-06-27 04:28:15,713] FATAL [KafkaServer id=2] Fatal error during KafkaServer startup. Prepare to shutdown (kafka.server.KafkaServer)
kafka.common.KafkaException: Socket server failed to bind to 10.0.165.8:9092: Cannot assign requested address.at kafka.network.Acceptor.openServerSocket(SocketServer.scala:331)at kafka.network.Acceptor.<init>(SocketServer.scala:256)at kafka.network.SocketServer$$anonfun$startup$1.apply(SocketServer.scala:97)at kafka.network.SocketServer$$anonfun$startup$1.apply(SocketServer.scala:89)at scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:48)at kafka.network.SocketServer.startup(SocketServer.scala:89)at kafka.server.KafkaServer.startup(KafkaServer.scala:229)at io.confluent.support.metrics.SupportedServerStartable.startup(SupportedServerStartable.java:112)at io.confluent.support.metrics.SupportedKafka.main(SupportedKafka.java:58)
Caused by: java.net.BindException: Cannot assign requested addressat sun.nio.ch.Net.bind0(Native Method)at sun.nio.ch.Net.bind(Net.java:433)at sun.nio.ch.Net.bind(Net.java:425)at sun.nio.ch.ServerSocketChannelImpl.bind(ServerSocketChannelImpl.java:223)at sun.nio.ch.ServerSocketAdaptor.bind(ServerSocketAdaptor.java:74)at sun.nio.ch.ServerSocketAdaptor.bind(ServerSocketAdaptor.java:67)at kafka.network.Acceptor.openServerSocket(SocketServer.scala:327)... 9 more
[2020-06-27 04:28:15,715] INFO [KafkaServer id=2] shutting down (kafka.server.KafkaServer)
[2020-06-27 04:28:15,717] INFO [SocketServer brokerId=2] Shutting down (kafka.network.SocketServer)
[2020-06-27 04:28:15,718] INFO [SocketServer brokerId=2] Shutdown completed (kafka.network.SocketServer)
[2020-06-27 04:28:15,721] INFO Shutting down. (kafka.log.LogManager)
[2020-06-27 04:28:15,760] INFO Shutdown complete. (kafka.log.LogManager)
[2020-06-27 04:28:15,761] INFO Terminate ZkClient event thread. (org.I0Itec.zkclient.ZkEventThread)
[2020-06-27 04:28:15,762] INFO Session: 0x27297ff0225a5a9 closed (org.apache.zookeeper.ZooKeeper)
[2020-06-27 04:28:15,764] INFO EventThread shut down for session: 0x27297ff0225a5a9 (org.apache.zookeeper.ClientCnxn)
[2020-06-27 04:28:15,765] INFO [KafkaServer id=2] shut down completed (kafka.server.KafkaServer)
[2020-06-27 04:28:15,766] INFO [KafkaServer id=2] shutting down (kafka.server.KafkaServer)

自己copy了server.properties文件到各个节点没有修改下面的配置监听器的配置，应该指向节点本身的主机名和端口，我全部四台机器都指向了10.0.165.8，所以导致了只有节点1是正常的

advertised.listeners=PLAINTEXT://10.0.165.9:9092

5.2 Confluent schema-registry启动失败

[2020-06-27 16:09:39,872] WARN The replication factor of the schema topic _schemas is less than the desired one of 3. If this is a production environment, it's crucial to add more brokers and increase the replication factor of the topic. (io.confluent.kafka.schemaregistry.storage.KafkaStore:242)
[2020-06-27 16:09:50,095] ERROR Server died unexpectedly:  (io.confluent.kafka.schemaregistry.rest.SchemaRegistryMain:51)
java.lang.IllegalArgumentException: Unable to subscribe to the Kafka topic _schemas backing this data store. Topic may not exist.at io.confluent.kafka.schemaregistry.storage.KafkaStoreReaderThread.<init>(KafkaStoreReaderThread.java:125)at io.confluent.kafka.schemaregistry.storage.KafkaStore.init(KafkaStore.java:130)at io.confluent.kafka.schemaregistry.storage.KafkaSchemaRegistry.init(KafkaSchemaRegistry.java:199)at io.confluent.kafka.schemaregistry.rest.SchemaRegistryRestApplication.setupResources(SchemaRegistryRestApplication.java:64)at io.confluent.kafka.schemaregistry.rest.SchemaRegistryRestApplication.setupResources(SchemaRegistryRestApplication.java:42)at io.confluent.rest.Application.createServer(Application.java:157)at io.confluent.kafka.schemaregistry.rest.SchemaRegistryMain.main(SchemaRegistryMain.java:43)

因为kafkaserver没有启动

6 常用操作

（1）启动

confluent start

（2）查看日志文件目录

confluent current

（3）列出连接

confluent list connectors

（4）查看加载的连接器

confluent status connectors

[
"file-source"
]

（5）查看具体连接器状态

confluent status file-source