本篇文章主要介绍作业池中是如何进行流处理的。最常见的场景就是在物联网平台中，将感知层收集的数据，上传到消息队列服务器中，然后服务层通过使用相应的客户端去接收数据，最后将数据进行处理，保存入库。最终通过app，小程序，web等方式展示结果。最核心的流程就是如何去接收消息中间件的数据，接收之后又是如何高效的处理数据呢？如果有这样的一个平台，既能通过快速的配置接入中间件，又能对数据进行转换处理，持久化处理（数据库，es，hbase），就能完美的解决物联网的数据问题。这样一个复杂的物联网项目就转化为一个普通的互联网项目，只需要对数据进行展示即可。本章就以mqtt协议为例，充分展示kettle平台的流处理能力。
?一，准备工作
mqtt服务器搭建，可参考https://www.emqx.com/zh
安装成功后，如下图：
mqtt生产者发送数据：
编写生产者：转换文件编写如下
?启动生产者，连接mqtt服务器，如下：
二，kettle工具pdi订阅mqtt数据
mqtt订阅topic，接收数据
通过子转换，处理流数据，将接收到的数据，调用遥测数据rest接口，进行处理。
三，kettle平台处理mqtt数据
上传到基于kettle的数据处理平台，并启动生产者与消费者
?遥测数据rest接口：
?流数据处理日志：
四，总结
本篇文章主要介绍kettle平台是如何处理流数据，当然不仅仅限于mqtt，还支持kafka，rabbit，socket，websocket等形式。
若有什么疑问，可加qq893393953。