基于Spark SQL 技術的工業數據統計研究

胡志寶 陸會明

（華北電力大學控制與計算機工程學院，北京102206）

1 概述

工業生產過程中，我們希望所有生產環節都能夠安全高效進行，但是實際生產中，某些物理量可能會出現較大波動，使得生產過程不夠穩定甚至發生危險，所以在工業生產過程中，我們會對某些環節進行監測記錄，以便在出現特殊情況時及時采取措施來保證生產過程的安全。當前傳感器、多媒體、數據庫和無線網絡技術在工業生產中得到了廣泛應用，由此開發了工業生產監控系統、工業生產流程處理系統等，這些系統在運行中會積累海量的數據。通過對這些數據進行挖掘，我們可以知道生產過程是否穩定，哪些量容易超過限制，哪些環節需要重點關注，找出生產過程的薄弱點對方案進行改進。由此可見，數據的挖掘分析可以提高工業生產決策的準確度，進一步改進工業生產效率[1]。

Spark[2]是專為大數據處理而設計的快速通用的計算引擎，是一個基于內存計算的可擴展的開源集群計算系統，解決了MapReduce[3]在大量的網絡傳輸和磁盤I/O 時效率低的問題。Spark 具有運行速度快、易用性、通用性、兼容性強的四大特點[4]。Spark SQL[5]是Spark 用來處理結構化數據的一個模塊，它提供了一個編程抽象叫做DataFrame，并且具有作為分布式SQL查詢引擎的作用，其使用類SQL 的語法作為高層的數據操縱API，極大地降低了數據分析工作的難度。

2 Spark SQL 技術介紹

2.1 Spark SQL 的功能

Spark SQL 是Spark 框架的重要組成部分，主要用于結構化數據處理和對Spark 數據執行類SQL 的查詢，其數據處理功能可以從以下三個過程來體現：

2.1.1 抽 取（Extract）：Spark SQL 可 以 從 多 種 文 件 系 統（HDFS[6]、S3. 本地文件系統等）、關系型數據庫（MySQL、Oracle等）或NoSQL 數據庫（Cassandra、HBase[7]、Druid 等）中獲取數據，Spark SQL 支持的文件類型可以是CSV、JSON、XML 等。

2.1.2 轉換（Transform）：在數據清洗方面，比如空值處理、拆分數據、規范化數據格式、數據替換等，Spark SQL 能夠準確高效地完成這類轉換操作。

2.1.3 加載（Load）：在將數據處理完成之后，Spark SQL 還可以將數據存儲到各種數據源中。

除了上述功能，SparkSQL 還可以和Spark 的其他模塊搭配使用，完成各種各樣更為復雜的工作，比如和SparkStreaming[8]搭配處理實時的數據流，和MLlib[9]搭配完成一些機器學習的應用[10]。Spark SQL 無縫地將SQL 查詢與Spark 程序混合，允許開發人員將結構化數據作為Spark 中的分布式數據集進行查詢，在Python，Scala 和Java 中集成了API，使得開發人員可以輕松地運行SQL 查詢以及復雜的分析算法。

2.2 Spark SQL 執行過程

Spark SQL 對SQL 語句的處理過程和關系型數據庫是類似的，會先對SQL 語句進行解析形成Tree，然后使用Rule 對Tree進行綁定、優化等處理，通過模式匹配對不同類型的節點采用不同的操作。Spark SQL 模塊劃分為Core、Catalyst、Hive[11]和Hive-ThriftServer 四大模塊，每個模塊的作用如下：

2.2.1 Core：負責處理數據的輸入/輸出，從不同的數據源獲取數據，然后將查詢結果輸出成DataFrame。

2.2.2 Catalyst：負責處理查詢語句的整個過程，包括解析、綁定、優化、物理計劃等，是Spark SQL 最為核心的部分，其性能優劣將決定整體的性能。

2.2.3 Hive：負責對Hive 數據的處理。

2.2.4 Hive-ThriftServer：提供CLI 和JDBC/ODBC 接口等。

圖1 Spark SQL 執行過程

圖2 讀取的數據

這四個模塊共同工作，完成數據的獲取和SQL 語句的執行任務，圖1 所示即為Spark SQL 執行SQL 語句的過程。

3 基于Spark SQL 的工業數據統計

對于得到的工業生產中的數據，通過對數據進行處理挖掘，得到我們需要的統計量，常見的統計量有均值、均方差、極值等。超限統計也是非常重要的一項統計內容，包括統計各個超限時間段及超限的持續時間。

3.1 數據源讀取

本文采用的是Java+ Spark SQL 的方式來實現數據的讀取與處理，數據讀取需要先創建SparkSession 服務，然后使用SparkSession 服務讀取數據。數據的第一列是日期列，讀取時將日期轉化為時間戳的形式，單位是毫秒。數據讀取的代碼如下:

圖2 為數據讀取結果。

timestamp 是時間列，對應著獲得每個數據的采樣時間點，其余列是在采樣時間點各個過程量的值。

3.2 統計的內容

3.2.1 常規統計量

常規統計量包括均值、均方差、最大值、最小值。通過均值和均方差我們可以知道過程量是否穩定在我們設置的輸入附近，生產過程是否穩定。這些統計量可以直接調用API 函數即可實現，過程如圖3。

圖3 常規統計流程

獲取常規統計量時與時間列無關且對時間列進行常規統計并沒有實際意義，所以先去掉時間列，然后對剩余列進行常規統計，實現的代碼如下，

data = data.drop("timestamp");

data = data.describe()。

3.2.2 超限統計

生產過程中一些過程量會有對應的上限值，過程量超過上限值時系統就會報警，就需要進行調節使過程量回歸正常值，一旦超限時間持續太長就會出現危險。超限統計就是對這種情況進行統計，統計超過限定值的持續時間以及最大超限時長。實際生產過程中會設置死區值，即過程量的值從超限狀態回歸正常值時，如果僅是下降穿過設定的上限值并不算回歸正常，還需要繼續下降一定的量才算回歸正常，繼續下降的這部分為死區。

超限統計牽扯到了時間段的統計，難點在于如何確定每一段超限時間段的起始時間點和終止時間點，只要得到起始時間點和終止時間點就可以計算出持續時長?？梢韵炔豢紤]死區的存在，僅考慮上限值，找出所有超過上限值的數據，然后再判斷這些數據相鄰行之間對應的時間點在時間上是否以采樣周期等距連續的，從而找出每一段連續的超限數據，記錄下來起始時間點。然后將上限值減去死區值得到的值作為新的上限值，以相同方法找出每一段超限數據，記錄下來結束時間點。將兩次記錄下來的時間點進行比較，可以得到考慮死區時每一段超限時間的起始時間點和結束時間點。解決步驟如下：

（1）將統計的數據與上限值進行比較，取出所有超過上限值的數據。

（2）判斷這些數據對應的時間點是否以采樣周期等距連續，找出連續的各個時間段。

（3）記錄找出的各個時間段的起始時間點，用列表list1 保存下來。

（4）上限值減去死區值作為新的上限值，重復（1）、（2）得到新的連續時間段，然后判斷得到的各個時間段內的數據是否存在超限數據，保留存在超限數據的時間段，記錄保留的時間段的結束時間點，用列表list2 保存下來。

（5）將兩個列表中的元素進行比較，確定在考慮死區時的每一段超限時間段的起始時間點列表和終止時間點列表，從而得到超限時間段。

得到超限時間段之后，即可計算出每段超限時間段的持續時長，通過比較得到最大時長，通過求和求出總的超限時長。超限統計流程如圖4 所示。

圖5 常規統計結果

圖4 超限統計流程

超限統計的代碼主要包括列表list1 和列表list2 的獲取，列表的獲取分別編寫了一個函數來實現。

（1）獲取list1 的函數及其參數如下:

public List＜List＜Long＞＞limit(Dataset data, Float set, String tag, Long T, SparkSession spark)；

函數的參數依次分別是被統計的數據、設定值、被統計的列的名稱、采樣周期、SparkSession 服務。返回值為步驟（2）中各個時間段的開始時間點組成的列表和結束時間點組成的列表，list1 為開始時間點組成的列表。函數的主要代碼語句與說明如下，

data = spark.sql (“select timestamp from tempTable where ”+tag+“＞”+set+“”); //取出超過設定值的數據

data = spark.sql(“select *, timestamp - lag(timestamp,1)over(order by timestamp) as diff from tempTable”); //將取出來的數據對應的采樣時間點相鄰行做差

data = spark.sql(“select timestamp,(case when diff = ”+T+“then 1 else 0 end) as status from tempTable”); //若上一步求得的差值等于采樣周期，則標記該行狀態為否則為0，生成狀態列data = spark.sql(“select timestamp,status,lead(status)over(order by timestamp) as lead from tempTable”); //將狀態列向前移動一行得到新的狀態列

根據兩列狀態列找出各個連續時間段的開始時間點和結束時間點

Dataset starttime = data.select(“timestamp”).filter(“status = 0 and lead = 1”);//開始時間表

Dataset endtime = data.select(“timestamp”).filter(“status = 1 and lead = 0”);//結束時間表

獲得的開始時間表和結束時間表是dataframe，還需要將其轉換為列表，開始時間表轉化的列表即為list1。

（2）獲取list2 的函數及其參數說明如下:

public List＜Long＞death(Dataset data, Float set, Float value,String tag, Long T, SparkSession spark)

函數的參數依次分別是被統計的數據、設定值、設定值減去死區值得到的數值，被統計的列的名稱、采樣周期、SparkSession服務。返回值分別為步驟（3）中保留的各個時間段的結束時間點組成的列表，即為list2 的值。函數主要代碼語句與說明如下：

List＜Long＞s = limit(data,value,tag,T,spark).get(0); //調用limit函數求出各個連續時間段的開始時間點

List＜Long＞e = limit (data,value,tag,T,spark).get (1); //調用limit 函數求出各個連續時間段的開始時間點

確定各個時間段是否存在超限數據，保留存在超限數據的時間段的結束時間點的集合作為返回值，即為list2 的值。

要想求出考慮死區情況的超限時間段，還需要將已經求出的list1 和list2 進行比較。list2 中的元素就是最終要求的超限時間段的結束時間點，然后將list1 中的元素與list2 中的元素依次比較來確定超限時間段的開始時間點。在得到各個超限時間段的開始和結束時間點之后，將列表轉換為DataFrame，開始列與結束列做差得到持續時長。

3.3 統計的結果

常規統計的結果如圖5 所示,超限統計選擇的是DOMAIN1:UNITMW 列，設定值為520，死區值為1，統計結果如圖6 所示。

圖6 超限時間段及其持續時長

4 結論

本文利用大數據處理技術Spark SQL 對數據進行統計，不僅統計了常用的一些統計量如均值、最值、均方差等，還進行了超限統計，有助于我們了解統計的數據所在的時間段內工業生產的狀態是否穩定，有沒有按照預期進行工作。對于得到的工業生產的數據，我們還可以利用這些數據進行建模，然后對模型進行優化、提高生產效率。同時Spark SQL 技術在數據處理方面有著很大的優勢，計算速度快、通用性高，還可以用于數據管理查詢系統。