大数据NiFi（二十一）：监控日志文件生产到Kafka

监控日志文件生产到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节点写出消息后，FlowFile将被路由到成功，而不需要等待响应。这提供了最好的性能，但可能会导致数据丢失。例如：消息写出到Kafka节点，但是对应节点挂掉，这时将消息路由到成功。

Use Transactions（使用事务）

“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”路由关系为自动终止。