大数据NiFi(二十一):监控日志文件生产到Kafka

举报
Lansonli 发表于 2023/03/05 11:09:11 2023/03/05
【摘要】 监控日志文件生产到Kafka案例:监控某个目录下的文件内容,将消息生产到Kafka中。此案例使用到“TailFile”和“PublishKafka_1_0”处理器。一、​​​​​​​配置“TailFile”处理器创建“TailFile”处理器并配置:注意:以上需要在NiFi集群中的每个节点上创建“/root/test/logdata”文件,“logdata”是文件,而非目录。二、配置“Pub...

监控日志文件生产到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需要将<Delivery Guarantee>属性设置为"Guarantee Replicated Delivery"。

  • 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需要将<Delivery Guarantee>属性设置为"Guarantee Replicated Delivery"。

“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/logdata


3、查看Kafka中自动创建的“nifi_topic”中的数据

以上数据每写入一行,有个空行,这是由于“TailFile”处理器监控数据导致的,实际就是写入了3条数据,可以通过后期业务处理时,对数据进行trim处理即可。

【版权声明】本文为华为云社区用户原创内容,转载时必须标注文章的来源(华为云社区)、文章链接、文章作者等基本信息, 否则作者和本社区有权追究责任。如果您发现本社区中有涉嫌抄袭的内容,欢迎发送邮件进行举报,并提供相关证据,一经查实,本社区将立刻删除涉嫌侵权内容,举报邮箱: cloudbbs@huaweicloud.com
  • 点赞
  • 收藏
  • 关注作者

评论(0

0/1000
抱歉,系统识别当前为高风险访问,暂不支持该操作

全部回复

上滑加载中

设置昵称

在此一键设置昵称,即可参与社区互动!

*长度不超过10个汉字或20个英文字符,设置后3个月内不可修改。

*长度不超过10个汉字或20个英文字符,设置后3个月内不可修改。