入门篇(二)
- 创建多个集群(broker)
- 同样的,先创建一个终端1启动zookeeper, 在创建一个终端2 cd 到kafka根目录
9. 设置多个broker集群
cp /usr/local/etc/kafka/server.properties /usr/local/etc/kafka/server-1.properties
cp /usr/local/etc/kafka/server.properties /usr/local/etc/kafka/server-2.properties
10. 编辑这些新建的文件,设置以下属性:
config/server-1.properties:
broker.id=1
listeners=PLAINTEXT://:9093
log.dir=/tmp/kafka-logs-1
config/server-2.properties:
broker.id=2
listeners=PLAINTEXT://:9094
log.dir=/tmp/kafka-logs-2
我们已经运行了zookeeper和刚才的一个kafka节点(server.properties),现在只需要再启动2个新的kafka节点。
- 分别新建不同的终端创建。
./bin/kafka-server-start /usr/local/etc/kafka/server-1.properties &
./bin/kafka-server-start /usr/local/etc/kafka/server-2.properties &
11. 创建一个新topic,把备份设置为:3
./bin/kafka-topics --create --zookeeper localhost:2181 --replication-factor 3 --partitions 1 --topic my-replicated-topic
怎么知道每个集群在做什么呢?运行命令“describe topics”
./bin/kafka-topics --describe --zookeeper localhost:2181 --topic my-replicated-topic
这是一个解释输出,第一行是所有分区(topics)的摘要,每一个线提供一个分区信息,因为我们只有一个分区,所有只有一条线。
"leader":该节点负责所有指定分区的读和写,每个节点的领导都是随机选择的。
"replicas":备份的节点,无论该节点是否是leader或者目前是否还活着,只是显示。
"isr":备份节点的集合,也就是活着的节点集合。
运行下面的命令,看看之前创建的节点topic
./bin/kafka-topics --describe --zookeeper localhost:2181 --topic test
没有惊喜,刚才创建的topic(主题/节点)没有Replicas,所以是0
- 除了手工创建topic外,你也可以配置你的broker,当发布一个不存在的topic时自动创建topic
发布一些信息到自己的集群server-1节点my-replicated-topic上
./bin/kafka-console-producer --broker-list localhost:9092 --topic my-replicated-topic
现在消费这些信息
./bin/kafka-console-consumer --zookeeper localhost:2181 --from-beginning --topic my-replicated-topic
关闭生产者,关闭消费者
control+c
我们要测试集群的容错,kill掉leader,Broker1作为当前的leader,也就是kill掉Broker1
ps | grep server-1.properties
7564 ttys002 0:15.91 /System/Library/Frameworks/JavaVM.framework/Versions/1.6/Home/bin/java...
使用 Kafka Connect (运行连接器)来 导入/导出 数据
- 1 、 首先,我们首先创建一些种子数据用来测试
cd /usr/local/Cellar/kafka/0.10.2.0
echo -e "foo\nbar" > test.txt
// 在test.txt增加一行
echo "Another line" >> test.txt
- 2 、 接下来,开始2个连接器运行在独立的模式,这意味着它们运行在一个单一的、本地的、专用的进程。我们提供3个配置文件作为参数。第一个始终是kafka Connect进程,如kafka broker连接和数据库序列化格式,剩下的配置文件每个指定的连接器来创建,这些文件包括一个独特的连接器名称,连接器类来实例化和任何其他配置要求的。
./bin/connect-standalone /usr/local/etc/kafka/connect-standalone.properties /usr/local/etc/kafka/connect-file-source.properties /usr/local/etc/kafka/connect-file-sink.properties
这是示例的配置文件,使用默认的本地集群配置并创建了2个连接器:第一个是导入连接器,从导入文件中读取并发布到Kafka主题,第二个是导出连接器,从kafka主题读取消息输出到外部文件,在启动过程中,你会看到一些日志消息,包括一些连接器实例化的说明。一旦kafka Connect进程已经开始,导入连接器应该读取从
test.txt
写入到connect-test
,导出连接器从主题connect-test
读取消息写入到test.sink.txt
通过验证输出文件的内容来验证数据数据已经全部导出:
《不需要启动kafka服务》
cat test.sink.txt
**
注意:`**,导入的数据也已经在Kafka主题
connect-test
里,所以我们可以使用该命令查看这个主题:
./bin/kafka-console-consumer --zookeeper localhost:2181 --topic connect-test --from-beginning
使用kafka steam 来处理数据(输入/读取数据)
./bin/kafka-topics --create \
--zookeeper localhost:2181 \
--replication-factor 1 \
--partitions 1 \
--topic streams-file-input
./bin/kafka-console-consumer --zookeeper localhost:2181 --from-beginning --topic streams-file-input
- 运行workCount来处理输入的数据
./bin/kafka-run-class org.apache.kafka.streams.examples.wordcount.WordCountDemo