在使用Kafka Connect时,需要注意一些事项,以帮助你构建适应长期需求的datapipeline。本章旨在提供有关的一些上下文。
要开始使用Kafka Connect,只有一个硬性的先决条件:一个Kafka的broker集群。然而,随着集群增长,有几个问题需要提前考虑:
在开始之前,确定哪种模式最适合您的环境非常有用。 对于适合单个代理的环境(例如从web服务器向Kafka发送日志),standalone模式非常适合。在单个source或sink可能需要大量数据的用例中(例如,将数据从Kafka发送到HDFS),分布式模式在可伸缩性方面更加灵活,并提供了高可用性服务,从而最小化停机时间。
Kafka Connect插件是一组jar文件,Kafka Connect可以在其中找到一个或多个connector、transform、以及converter的实现。Kafka Connect将每个插件彼此隔离,这样一个枣枯插件中的库就不会受到其他插件库的影响,这点非常重要旁罩。
Kafka Connect plugin是:
(1)在一个uber jar文件中包含插件及所有第三方依赖;或
(2)一个包含jar包和第三方依赖的目录。
Kafka Connect使用plugin path找到插件,这是Kafka Connect在worker配置文件中定义的一个以逗号分隔的目录列表。要安装插件,请将目录或uber jar放在plugin path路径中列出的目录中。
举个例子 ,我们在每台机器上创建一个/usr/local/share/kafka/plugins目录,然后将我们所有的插件jar或插件目录放入其中。然后在worker的配置文件中加入如下配置项:
现在,当我们启动worker时,Kafka Connect可以发现这些插件中定义的所有connector、transform以及converter。Kafka Connect显式地避免了其他插件中的库, 并防止了冲突。
如果要在同一个机器上运行多个standalone实例,有一些参数需要是独一无二的:
(1)offset.storage.file.filename:connector偏移量的存储。
(2)rest.port:用于监听http请求的rest接口所占用的端口。
connector和task的配置,offsets和状态会存储在Kafka的内部主题中,Kafka Connect会自动创建这些主题,且所有topic都使用了压缩清理策略。
如果要手动创建这些topic,推荐使用如下命令:
这里只列出一些有疑问的。
配置了group.id的worker会自动发现彼此并形成集群。一个集群中的所有worker必须使用相同的三个Kafka topic来共享配置、偏移量以及状态,所有worker必须配置相同的config.storage.topic、offset.storage.topic以及status.storage.topic。
每个converter实现类都有自运岩闹己的相关配置需求。下面的例子展示了一个worker属性文件,其中使用的AvroConverter需要将Schema Registry的url作为属性进行传递。
注意: 除了其配置覆盖这些配置的connector,worker上运行的所有connector都使用这些converter。
1、安装
brew install kafka
会一喊皮知起安装zookeeper和Kafka两个插件
2、文件路径
(1)Kafka的bin目录路径
/usr/local/Cellar/kafka/2.5.0/bin,里面包含Kafka的各种 *** 作命令
kafka-server-start,服务启动命令
kafka-server-stop,服务停止命令
kafka-topics,主题 *** 作命令
kafka-console-producer,生产消息命令
kafka-console-consumer,消费消息
(2)Kafka的配置文件路径
/握贺usr/local/etc/kafka
zookeeper.properties,zookeeper的配置文件
server.properties,Kafka服务的配置文件
3、必须的前提修改Kafka的配置文件,否则后面的启动会失败
vim /usr/local/etc/kafka/server.properties
增加:listeners=PLAINTEXT://localhost:9092
其中有改行,默认被注释掉了,打开修改即可
4、打开新终端窗口1,进入Kafka的bin目录启动zookeeper服务
kafka-server-start /usr/local/etc/kafka/server.properties &
都是info,没有报fatal或者error错误,即为启动成功。
如果不先修改第3步的配置文件,那么会报错如下
ERROR Exiting Kafka due to fatal exception (kafka.Kafka$)
5、打开新终端窗口2,进入Kafka的bin目录启动Kafka资深的服务
kafka-server-start /usr/local/etc/kafka/server.properties &
都是info,没有报fatal或者error错误,即为启动成功。
如果不先修改第3步的配置文件,那么会报错如下
ERROR Exiting Kafka due to fatal exception (kafka.Kafka$)
6、打开新终端窗口3,进入Kafka的bin目录,新建topic主题test
kafka-topics --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test
出现:Created topic test. 表明建立topic成功
查看所有目前的topic,校验是否新建完毕
kafka-topics --list --zookeeper localhost:2181
会列出所有建立的topic
7、打开新终端窗口4,进入Kafka的bin目录,郑消打开生产者客户端,topic为test的主题
kafka-console-producer --topic test --broker-list localhost:9092
出现>表示启动生产者客户端成功,等待输入生产者消息,输入消息,回车,即表示插入了一条消息到主题test的队列中
8、打开新终端窗口5,进入Kafka的bin目录,打开消费者客户端,topic为test的主题
kafka-console-consumer --bootstrap-server localhost:9092 -topic test
等待显示生产者输入的消息,并同步显示出来
欢迎分享,转载请注明来源:内存溢出
评论列表(0条)