基于Spark的物联网设备故障实时检测与分析

首页 » 物联网 » 基于Spark的物联网设备故障实时检测与分析

基于Spark的物联网设备故障实时检测与分析

  • 一、业务场景
  • 二、数据集说明
  • 三、操作步骤
    • 阶段一、启动HDFS、Spark集群服务
    • 阶段二、准备案例中用到的数据集
    • 阶段三、处理实时数据
    • 阶段四、自行练习

未经许可,禁止以任何形式转载,若要引用,请标注链接地址

全文共计3770字,阅读大概需要3分钟

一、业务场景

通过实时计算所有计算机机架在一个滑动窗口上的平均温度,从而发现温度异常的机架,及时检测故障的发生。

二、数据集说明

本案例用到的实时数据说明如下:
  - 原始数据文件位于本地/data/dataset/streaming/iot/file1.json和/data/dataset/streaming/iot/file2.json

以下为数据中心的两个数据传感器检测到的两个机架的温度数据:
  file1.json:

   {“rack”:”rack1”,”temperature”:99.5,”ts”:”2017-06-02T08:01:01”}
   {“rack”:”rack1”,”temperature”:100.5,”ts”:”2017-06-02T08:06:02”}
   {“rack”:”rack1”,”temperature”:101.0,”ts”:”2017-06-02T08:11:03”}
   {“rack”:”rack1”,”temperature”:102.0,”ts”:”2017-06-02T08:16:04”}

file2.json:

   {“rack”:”rack2”,”temperature”:99.5,”ts”:”2017-06-02T08:01:02”}
   {“rack”:”rack2”,”temperature”:105.5,”ts”:”2017-06-02T08:06:04”}
   {“rack”:”rack2”,”temperature”:104.0,”ts”:”2017-06-02T08:11:06”}
   {“rack”:”rack2”,”temperature”:108.0,”ts”:”2017-06-02T08:16:08”}

三、操作步骤

阶段一、启动HDFS、Spark集群服务

1、启动HDFS集群
  在Linux终端窗口下,输入以下命令,启动HDFS集群:

1.	$ start-dfs.sh

2、启动Spark集群
  在Linux终端窗口下,输入以下命令,启动Spark集群:

1.	$ cd /opt/spark
2.	$ ./sbin/start-all.sh

3、验证以上进程是否已启动
  在Linux终端窗口下,输入以下命令,查看启动的服务进程:

1.	$ jps

如果显示以下5个进程,则说明各项服务启动正常,可以继续下一阶段。

1.	2288 NameNode
2.	2402 DataNode
3.	2603 SecondaryNameNode
4.	2769 Master
5.	2891 Worker

阶段二、准备案例中用到的数据集

1、在HDFS上创建数据存储目录。在Linux终端窗口下,输入以下命令:

1.	$ hdfs dfs -mkdir -p /data/dataset/streaming/iot-input

2、将iot数据文件上传到HDFS上。在Linux终端窗口下,输入以下命令:

1.	$ hdfs dfs -put /data/dataset/streaming/iot/*  /data/dataset/streaming/iot-input/

3、在Linux终端窗口下,输入以下命令,查看HDFS上是否已经有了iot目录:

1.	$ hdfs dfs -ls /data/dataset/streaming/iot-input

这时应该看到该目录下有两个文件:file1.json和file2.json。

阶段三、处理实时数据

1、启动spark shell。在Linux终端窗口中,输入以下命令:(注意,请将以下命令中的localhost替换为虚拟机实际的机器名)

1.	$ spark-shell --master spark://localhost:7077

执行以上代码,进入到Spark Shell交互开发环境:

基于Spark的物联网设备故障实时检测与分析

2、首先需要导入案例代码运行所依赖的包。在paste模式下,输入以下代码:

1.	// 首先导入包
2.	import org.apache.spark.sql.types._
3.	import org.apache.spark.sql.functions._

同时按下【Ctrl + D】按键,执行以上代码,输出结果如下:
基于Spark的物联网设备故障实时检测与分析

3、构造数据Schema模式,注意其中温度和时间的数据类型。在paste模式下,输入以下代码:

1.	// 定义schema
2.	val iotDataSchema = new StructType().add("rack", StringType, false)
3.	                                    .add("temperature", DoubleType, false)
4.	                                    .add("ts", TimestampType, false)

同时按下 【Ctrl + D】 键,执行以上代码,输出结果如下:

基于Spark的物联网设备故障实时检测与分析

4、加载数据文件到DataFrame中。在paste模式下,输入以下代码:

1.	// 读取温度数据
2.	val dataPath = "hdfs://localhost:9000/data/dataset/streaming/iot-input"
3.	val iotSSDF = spark.readStream.schema(iotDataSchema).json(dataPath)

同时按下 【Ctrl + D】 键,执行以上代码,输出内容如下:
基于Spark的物联网设备故障实时检测与分析

由以上输出内容可以看出,采集到的机架温度(json格式)数据被读取到了DataFrame中。

5、创建10分钟大小的滑动窗口,并在temperature列上求机架温度的平均值。在paste模式下,输入以下代码:

1.	// group by一个滑动窗口,并在temperature列上求平均值
2.	val iotAvgDF = iotSSDF.groupBy(window($"ts", "10 minutes", "5 minutes"))
3.	                      .agg(avg("temperature") as "avg_temp")

同时按下 【Ctrl + D】 键,执行以上代码,输出内容如下:

基于Spark的物联网设备故障实时检测与分析

6、将数据写出到memory data sink。在paste模式下,输入以下代码:

1.	// 将数据写出到memory data sink,使用查询名称iot
2.	val iotMemorySQ = iotAvgDF.writeStream
3.	                          .format("memory")
4.	                          .queryName("iot")
5.	                          .outputMode("complete")
6.	                          .start()

同时按下 【Ctrl + D】 键,执行以上代码,输出内容如下:
基于Spark的物联网设备故障实时检测与分析

7、在iot上执行sql查询,显示数据,以start时间排序。在paste模式下,输入以下代码:

1.	// 显示数据,以start时间排序
2.	spark.sql("select * from iot").orderBy($"window.start").show(false)

同时按下【 Ctrl + D 】键,执行以上代码,输出结果如下:
基于Spark的物联网设备故障实时检测与分析

从实时检测的结果来看,机架的平均温度一直在上升,其中8:15-8:25这十分钟内的机架温度最高。

8、停止该流查询,命令行下执行如下命令:

1.	iotMemorySQ.stop

按下【Enter】键,执行以上代码。

阶段四、自行练习

1、在本案例的基础上,请继续分析,找出是哪些机架的温度在上升。

本文章来源于互联网,如有侵权,请联系删除!

相关推荐: 物联网通信技术,那些你不知道的事

摘要:通信技术是物联网的基础,如果把物联网比作是物流系统,那么通信技术就相当于 是送快递的各种运输方式,比如空运、水运还有陆运等。在通信技术当中,大体上它可以分为两大类,一类是无线通信技术,另一类是有线通信技术。 本文分享自华为云社区《物联网通信技术之有线通信…