入门篇(二)
- 创建多个集群(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
集群server-1.properties的信息
这是一个解释输出，第一行是所有分区（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
网友评论