监控日志文件生产到Kafka
案例：监控某个目录下的文件内容，将消息生产到Kafka中。
(资料图)
此案例使用到“TailFile”和“PublishKafka_1_0”处理器。
一、配置“TailFile”处理器
创建“TailFile”处理器并配置：
注意：以上需要在NiFi集群中的每个节点上创建“/root/test/logdata”文件，“logdata”是文件，而非目录。
二、配置“PublishKafka_1_0”处理器
“PublishKafka_1_0”处理器作用是使用Kafka 1.0生产者API将FlowFile的内容作为消息发送给Apache Kafka。发送的内容可以是单独的FlowFile，也可以通过用户指定分隔符分割的FlowFile内容。
关于“PublishKafka_1_0”处理器的“Properties”主要配置的说明如下：
配置项 | 默认值 | 允许值 | 描述 |
|---|---|---|---|
Kafka Brokers（Kafka节点） | localhost:9092 | 逗号分割的Kafka集群Broker列表。格式：host:port | |
Topic Name（topic 名称） | 将消息生产到的Topic 名称。 | ||
Delivery Guarantee（数据传递保证） | 0 | 指定保证消息被发送到Kafka的要求。对应Kafka的"acks"属性。可以配置的项如下：Best Effort （尽力交付，相当于ack=0）:在向Kafka节点写出消息后，FlowFile将被路由到成功，而不需要等待响应。这提供了最好的性能，但可能会导致数据丢失。例如：消息写出到Kafka节点，但是对应节点挂掉，这时将消息路由到成功。Guarantee Single Node Delivery(保证单节点交付，相当于ack=1，Kafka中的默认配置):KafkaProducer把消息发送出去，至少要等待leader已经成功将数据写入本地log，但是并没有等待所有follower是否成功写入。该情况下，如果follower没有成功备份数据，而此时leader刚好又挂掉了，就会导致消息丢失。该选项就是如果消息被单个Kafka节点接收到，FlowFile将被路由到成功，无论它是否被复制，但如果Kafka节点崩溃，可能会导致数据丢失。Guarantee Replicated Delivery(保证复制交付，相当于ack=-1):FlowFile数据写出后，Kafka topic ISR列表离跟leader保持同步的那些follower都要把消息同步过去,该消息才会被认为成功，否则路由到失败。 | |
Use Transactions（使用事务） | true | ▪true▪false | 指定NiFi是否应该在与Kafka通信时提供事务性保证。如果发送数据到Kafka有问题，并且这个属性设置为false，那么已经发送到Kafka的消息将继续发送，并被传递给消费者。如果这个设置为true，那么Kafka事务将被回滚，这样这些消息对消费者是不可用的。将此设置为true需要将 |
在向Kafka节点写出消息后，FlowFile将被路由到成功，而不需要等待响应。这提供了最好的性能，但可能会导致数据丢失。例如：消息写出到Kafka节点，但是对应节点挂掉，这时将消息路由到成功。
Guarantee Single Node Delivery(保证单节点交付，相当于ack=1，Kafka中的默认配置):KafkaProducer把消息发送出去，至少要等待leader已经成功将数据写入本地log，但是并没有等待所有follower是否成功写入。该情况下，如果follower没有成功备份数据，而此时leader刚好又挂掉了，就会导致消息丢失。该选项就是如果消息被单个Kafka节点接收到，FlowFile将被路由到成功，无论它是否被复制，但如果Kafka节点崩溃，可能会导致数据丢失。 Guarantee Replicated Delivery(保证复制交付，相当于ack=-1): FlowFile数据写出后，Kafka topic ISR列表离跟leader保持同步的那些follower都要把消息同步过去,该消息才会被认为成功，否则路由到失败。 Use Transactions（使用事务）true true false 指定NiFi是否应该在与Kafka通信时提供事务性保证。如果发送数据到Kafka有问题，并且这个属性设置为false，那么已经发送到Kafka的消息将继续发送，并被传递给消费者。如果这个设置为true，那么Kafka事务将被回滚，这样这些消息对消费者是不可用的。将此设置为true需要将
“PublishKafka_1_0”处理器配置如下：
1、创建“PublishKafka_1_0”处理器
2、配置“PROPERTIES”
注意：以上topic 可以在Kafka中创建好，也可以执行时自动创建。
3、连接“TailFile”处理器和“PublishKafka_1_0”处理器
连接“TailFile”处理器和“PublishKafka_1_0”处理器，并设置“PublishKafka_1_0”处理器“failure”和“success”路由关系为自动终止。
三、运行测试
1、启动Kafka集群，启动NiFi处理流程
2、向/root/test/logdata文件中写入数据并保存
向NiFi集群中的其中一台节点的“logdata”中写入以下数据即可
[root@node1 test]# echo "hello world1" > /root/test/logdata[root@node1 test]# echo "hello world2" >> /root/test/logdata[root@node1 test]# echo "hello world3" >> /root/test/logdata3、查看Kafka中自动创建的“nifi_topic”中的数据
以上数据每写入一行，有个空行，这是由于“TailFile”处理器监控数据导致的，实际就是写入了3条数据，可以通过后期业务处理时，对数据进行trim处理即可。
关键词： Kafka
