使用ZeroMq源测试Kuiper吞吐量

Posted 2023-05-12

参考技术A

EMQ X Kuiper支持各种source（MQTT、ZeroMq、EdgeX），默认支持MQTT
1、关于Kuiper的介绍及简单使用，请参考： https://docs.emqx.io/kuiper/latest/cn/
2、关于Kuiper与EdgeX Foundry集成，请参照： https://www.jianshu.com/p/0726d41b00bf

本文实践ZeroMq作为消息总线源（实际上EdgeX内部使用的正是ZeroMq），Kuiper订阅其消息，并以此测试Kuiper的性能，kuiper官方给出性能测试结果： <u>https://docs.emqx.io/kuiper/latest/cn/</u>

操作系统 ： Ubuntu 18.04
测试原理描述 ：启动一个go应用向ZeroMq发送消息，同时Kuiper订阅来自ZeroMq的消息，经过Kuiper规则处理后输出消息
操作步骤 ：
1、创建一个go应用，向ZeroMq发送若干条消息，每条消息类似：
"device":"demo","readings":["device":"Temperature device","name":"Temperature","value":"40","device":"Humidity device","name":"Humidity","value":"45"],大小大概为157字节。

go应用源码： https://github.com/emqx/kuiper/blob/master/fvt_scripts/edgex/benchmark/pub.go
go应用与kuiper在一台机器上
2、由于Kuiper对zmq源的支持是以插件的形式支持的，而且必须与Kuiper的版本相配套，故此处kuiper采用源码编译打包，同时编译zmq的插件

3、修改source配置

10.0.105.143是我的go应用所在的主机IP

4、启动Kuiper server进程

5、设置Kuiper订阅的数据流及过滤规则

Kuiper规则配置如下：

以上参数含义： https://github.com/emqx/kuiper/blob/master/docs/zh_CN/rules/overview.md

6、go应用发送消息耗时：0.344352s

同时kuiper侧订阅zeroMq主题为events的消息，过滤了所有消息

说明，kuiper在0.344352s处理了9676条数据

经测试：

注：测试发现当数据量大时，明显丢包严重

1. Spark Streaming概述

1.1 什么是Spark Streaming

　　　　　　

　　　　　　Spark Streaming类似于Apache Storm，用于流式数据的处理。根据其官方文档介绍，Spark Streaming有高吞吐量和容错能力强等特点。Spark Streaming支持的数据输入源很多，例如：Kafka、Flume、Twitter、ZeroMQ和简单的TCP套接字等等。数据输入后可以用Spark的高度抽象原语如：map、reduce、join、window等进行运算。而结果也能保存在很多地方，如 HDFS，数据库等。另外Spark Streaming也能和MLlib(机器学习)以及Graphx完美融合

　　　　　　

　　　　　　和Spark基于RDD的概念很相似，Spark Streaming使用离散化流(discretized stream)作为抽象表示，叫作DStream。DStream是随时间推移而收到的数据的序列。在内部，每个时间区间收到的数据都作为RDD存在，而DStream是由这些RDD所组成的序列(因此得名“离散化”)

　　　　　　

1.2 为什么要学习Spark Streaming 

　　　　　　1. 易用

　　　　　　

　　　　　　2. 容错

　　　　　　

　　　　　　3. 易整合到Spark体系

　　　　　　

1.3 Spark与Storm的对比

Spark	Storm
开发语言:Scala	开发语言:Clojure
编程模型:DStream	编程模型:Spout/Bolt