基于Oracle Streaming Service的IoT数据实时分析

5G时代的来临意味着物联网行业将会有飞跃性的发展，大量的传感器每时每刻都在产生着大量的数据。对应用物联网（IoT）数据进行有效的处理及分析显得越来越重要。

既然IoT数据分析这么火，今天我就和大家分享下，年前做的一个IoT数据实时分析的Case，一起入个门！

由于信息安全问题，企业相关信息不做详细介绍。企业方的需求是从病房中的呼吸心跳传感器，以及空气传感器进行实时数据采集，并按照业务逻辑进行数据加工，对加工后的数据搭建预测模型，最后通过前端报表工具对预测结果及实时数据进行可视化展示。

基于上述业务需求，我们来看下技术架构！

架构说明

• 在一台虚拟机上搭建了MQTT Broker服务，用以接收IoT传感器的实时数据。目前是单节点的虚拟机，考虑到生产系统的HA（高可用），后续会切换为集群或基于k8s搭建

• 在一台虚拟机上部署k8s环境，并搭建MQTT Connecotor和Kafka Connector，用以连接MQTT Broker和 Oracle Streaming Service，把实时数据加载到Streaming Service队列中

• 通过Data Loader（部署在k8s环境中），把数据实时加载到数仓ADW（Autonomous Data Warehouse），这里数据加载可以用Oracle Funtions进行替代（类似于AWS Lambda）可运行代码来触发事件并为自动管理底层计算资源

• 在ADW中进行数据处理加工和预测模型的搭建，并通过前端可视化工具OAC（Oracle Analytics Cloud）对计算结果及实时数据，进行实时可视化展示
  

可能有的童鞋会想，为什么不直接把IoT数据加载到Streaming Service，中间要加这么多层？

主要是因为，目前传感器由于轻量化的原因，大多是基于C语言开发的。而绝大部分的Queue Service不直接提供C语言接口，所以在 Streaming Service之前，搭建了中间层做连接

系统实现

在Oracle Cloud上实现上述架构，可分为8个部分！

1. kafka-connect-mqtt 部署

2. Streaming Service 部署

3. Oracle Kubernetes Engine 部署

4. OKE中部署Docker

5. Hive MQ 部署

6. Stream数据加载ADW

7. ADW数据加工及机器学习模型搭建

8. OAC前端可视化界面开发

由于内容较多，为了方便理解，我会每周分批进行介绍。

今天就先给大家介绍下第一部分！

kafka-connect-mqtt 部署

本步骤中需要在Oracle Linux VM中部署 Docker环境，再从 Github上下载 kafka-connect-mqtt Docker文件，最后把 Docker Image Push到 Oracle Cloud Container Registry上。

1. Policy设定

需要给Admin 用户添加Streaming Service的权限

路径：OCI Menu → Identity → Policies → Create Policy

allow group Administrators to manage streams in tenancy
allow group Administrators to manage stream-pull in tenancy
allow group Administrators to manage stream-push in tenancy

‍

2. 在 Oracle Linux VM中部署 Docker

生成一个Oracle VM

路径：OCI Menu → Instances → Create Instance, 参数默认, 添加SSH Public Key

VM生成完毕后的画面如下

连接VM，其中id_rsa为 Private Key文件

ssh -i id_rsa opc@130.61.53.6

修改selinux config文件，把selinux修改为disable

sudo vi /etc/selinux/config

修改selinux状态后重启，并确认状态

sudo shutdown -r now
sestatus

移动到repo目录，并安装 docker

cd /etc/yum.repos.d/
sudo wget http://yum.oracle.com/public-yum-ol7.repo
sudo yum -y install docker-engine


sudo systemctl start docker
sudo systemctl status docker

‍

sudo systemctl enable docker

配置docker权限后重启

docker ps
sudo usermod -aG docker $USER

3. kafka-connect-mqtt Docker部署

kafka-connect-mqtt下载

wget https://github.com/lensesio/stream-reactor/releases/download/1.2.3/kafka-connect-mqtt-1.2.3-2.1.0-all.tar.gz
mkdir -p kafka-connect-mqtt
tar -C kafka-connect-mqtt -xvf kafka-connect-mqtt-1.2.3-2.1.0-all.tar.gz

执行Docker，并建立目录

docker run -it confluentinc/cp-kafka-connect /bin/bash
mkdir -p /etc/kafka/plugins/lib

确认Container list，并复制刚刚生成的Container ID

docker ps --all

把kafka-connect-mqtt-1.2.3-2.1.0-all.jar复制到 刚刚生成的Container目录中

cd kafka-connect-mqtt/
$ docker cp kafka-connect-mqtt-1.2.3-2.1.0-all.jar fdea37d324a0:/etc/kafka/plugins/lib/kafka-connect-mqtt-1.2.3-2.1.0-all.jar

3. 上传到 Oracle Container Registry

把上面生成的Docker Image Push 到 Oracle Container Registry

# Docker Registry是根据 Region区分. Region名如下<region>.ocir.io
# Ex: fra Frankfurt, iad Ashburn, lhr London, phx Phoenix
# Username为 tenancy-namespace/oracleidentitycloudservice/<user-name>
# tenancy-namespace可在 OCI>Developer Service>Registry中 确认
# Password为 Authtoken




$ docker login fra.ocir.io
Username: fry97l7sa3hs/oracleidentitycloudservice/chulnam.kim@oracle.com
Password: )Md:UnOpM..rY8rxg-#V




# commit之前生成的 docker，然后Push到 Oracle Registry
$ docker commit fdea37d324a0 fra.ocir.io/fry97l7sa3hs/kafka-connect-mqtt/kafka-connect:0.1
$ docker push fra.ocir.io/fry97l7sa3hs/kafka-connect-mqtt/kafka-connect:0.1

在Oracle Registry确认上传的Image

路径：OCI Menu → Developer Services → Registry(OCIR)

第一部就到这里！

下期咱们接着看，如何部署Streaming Service和 Oracle Kubernetes Engine！