Apache Nifi挂接MQTT与Kafka实践

目录

1.  说明：

2. 方案设计：

2.1 资源配置：

2.2 交互Topics:  

3. 实现步骤 

3.1 Nifi 桌面

3.2 MqttToKafka

3.2.1 配置

3.2.2 测试

3.2.3 结果

3.3 KafkaToMqtt

3.3.1 配置 

3.3.1 测试

3.3.1 结果 ​编辑

4. 总结：

4.1 知识点

Nifi Kafka Processor 配置字典：

Topic通配符:

5. 参考：

1.  说明：

      在一些方案实现过程中，感觉需要一种接驳器来连接不同的数据源并汇流到一处进行统一处理，于是寻到NIFI（官网）这个工具，它相当于“数据水管+接驳器工具箱”，能丝滑联结不同的数据源，总体思路是把各类数据源汇流到Kafka集中处理，比如日志文件，消息传递，数据库操作等。初步使用感觉很不错，分享之。

2. 方案设计：

- 连接Emqx集群(mqtt服务)与Kafka集群，实现数据流动的双工运作 - 客户端(连mqtt) <=> 应用服务(连kafka)

2.1 资源配置：

简单起见，在docker环境中实施，后续迁移到K8s

Nifi的docker配置

# 建个卷，持久化数据
docker volume create nifi_data

docker-compose.yml

version: "3.7"
services:nifi:image: apache/nifi:1.9.2container_name: nifirestart: alwaysports:- "9080:8080"environment:- NIFI_WEB_HTTP_HOST=0.0.0.0#- NIFI_HOME=/home/nifi#- NIFI_LOG_DIR=/home/nifi/logsvolumes:- nifi_data:/home/nifivolumes:nifi_data:external: true

2.2 交互Topics:  

3. 实现步骤 

3.1 Nifi 桌面

配好后，访问​http://host001.dev.ia:9080/nifi/​, 中间是配好的两个Processor Group，分别是MqttToKafka与KafkaToMqtt，代表双向流动配置。

3.2 MqttToKafka

3.2.1 配置

 加ConsumeMQTT Processor：拉Processor组件下去，点开选ConsumeMQTT

加PublishKafka_2_0 Processor：拉Processor组件下去，点开选PublishKafka_2_0

拖动ConsumeMQTT连接PublishKafka, 会添加一个队列连接组件命名为 Message

正确运行如图：

3.2.2 测试

说明：

1. 用mqtt客户端工具MqttX向topic=tset.topic.nifi1发送json数据包
2. 用python脚本作为消费者客户端连接kafka，订阅topic=tset.topic.nifi1，获取该数据包

python脚本:

from confluent_kafka import Consumer, KafkaError, KafkaException
import asyncio
import jsonasync def consume_loop(consumer, topics):try:# 订阅主题consumer.subscribe(topics)while True:# 轮询消息msg = consumer.poll(timeout=1.0)if msg is None:continueif msg.error():if msg.error().code() == KafkaError._PARTITION_EOF:# End of partition eventprint("%% %s [%d] reached end at offset %d\n"% (msg.topic(), msg.partition(), msg.offset()))elif msg.error():raise KafkaException(msg.error())else:# 正常消息raw_message = msg.value()print(f"Raw message: {raw_message}")parsed_message = json.loads(raw_message.decode("utf-8"))print(f"Received message: {type(parsed_message)} : {parsed_message}")await asyncio.sleep(0.01)  # 小睡片刻，让出控制权finally:# 关闭消费者consumer.close()async def consume():# 消费者配置conf = {"bootstrap.servers": "host001.dev.ia:19092,host001.dev.ia:29092,host001.dev.ia:39092","group.id": "mygroup1","auto.offset.reset": "earliest",}# 创建消费者consumer = Consumer(conf)await consume_loop(consumer, ["tset.topic.nifi1"])if __name__ == "__main__":asyncio.run(consume())

3.2.3 结果

脚本 Nifi 

3.3 KafkaToMqtt

3.3.1 配置 

加ConsumeKafkaProcessor：拉Processor组件下去，点开选ConsumeMQTT

加PublishMQTT Processor：拉Processor组件下去，点开选PublishMQTT

 拖动ConsumeKafka_2_0连接PublishMQTT, 会添加一个队列连接组件命名为 Message

正确运行如图：

3.3.1 测试

说明：

1. python脚本向Kafka发布消息到 topic = test.topic.bus
2. MqttX客户端订阅接收

脚本

from confluent_kafka import Producer
import jsondef delivery_report(err, msg):"""Called once for each message produced to indicate delivery result.Triggered by poll() or flush()."""if err is not None:print(f"Message delivery failed: {err}")else:print(f"Message delivered to {msg.topic()} [{msg.partition()}]")def create_async_producer(config):"""Creates an instance of an asynchronous Kafka producer."""return Producer(config)def produce_messages(producer, topic, messages):"""Asynchronously produces messages to a Kafka topic."""for message in messages:# Trigger any available delivery report callbacks from previous produce() callsproducer.poll(0)# Asynchronously produce a message, the delivery report callback# will be triggered from poll() above, or flush() below, when the message has# been successfully delivered or failed permanently.producer.produce(topic, json.dumps(message).encode("utf-8"), callback=delivery_report)# Wait for any outstanding messages to be delivered and delivery report# callbacks to be triggered.producer.flush()if __name__ == "__main__":# Kafka configuration# Replace these with your server's configurationconf = {"bootstrap.servers": "host001.dev.ia:19092,host001.dev.ia:29092,host001.dev.ia:39092",  # Replace with your Kafka server addresses# "client.id": "python-producer",}# Create an asynchronous Kafka producerasync_producer = create_async_producer(conf)# Messages to send to Kafkamessages_to_send = [{"key": "value1"}, {"key": "value2"}, {"key": "value3"}]# Produce messagesproduce_messages(async_producer, "test.topic.bus", messages_to_send)

3.3.1 结果 

MqttX 

Nifi 

4. 总结：

       Nifi支持集群化部署，如此从数据采集，数据流动到数据存储都实现了分布式，而且有可视化的界面可方便地进行数据节点的集聚与增减配置，目前只是浅尝即止，更深入的研究待后续不断补充优化。

4.1 知识点

Nifi Kafka Processor 配置字典：

Topic通配符:

5. 参考：

- https://zhuanlan.zhihu.com/p/697301397

- https://blog.51cto.com/u_16213319/7344183

- Apache NiFi Docker Compose | All About

- https://blog.51cto.com/u_16099203/10959511

- 大数据NiFi（二十一）：监控日志文件生产到Kafka-腾讯云开发者社区-腾讯云

- PublishMQTT