讓你更聰明的石油和天然氣資產實施預見性維護磚

如何構建一個端到端的預測和磚三角洲和火花的流媒體數據管道嗎
在磚試試這個筆記本

維護資產,如壓縮機是一個極其複雜的努力:它們用於從小型鑽機到深水平台、資產遍布全球各地,他們每天生成tb的數據。Beplay体育安卓版本失敗隻是一個壓縮機成本每天數百萬美元的生產損失。節省時間和金錢的一個重要方法是使用機器學習來預測故障維修工作指令和問題之前發生故障。

最後,你需要建立一個端到端的預測數據管道,可以提供一個實時數據庫保持資產部分和傳感器映射,支持連續大量的遙測應用程序流程,並允許您對這些數據集預測壓縮機故障。

我們解決這些問題的方法是通過選擇一個統一的平台,提供了這些功能。Beplay体育安卓版本磚提供了一個統一的分析平台Beplay体育安卓版本彙集了大數據和人工智能的不同角色,並允許您的組織一起合作在一個工作區。磚的其他重要的優勢統一分析平台包括的能力:Beplay体育安卓版本

• 自旋向上必要的資源和數據科學家、工程師、數據和數據分析師快速理解數據。
• 有多重雲策略允許每個人都使用相同的協作工作區beplay娱乐iosAzure或AWS。
• 站起來一套不同的實例類型組合最優運行工作負載
• 調度命令(包括REST API命令),允許您自動創建和auto-terminate集群。
• 快速而方便地啟用訪問控製分配權限以及允許訪問令牌為安全的REST API調用當productionizing你的解決方案。

在這篇文章中,我們將向您展示如何讓你更聰明的石油和天然氣資產:

• 在磚使用火花流來處理大量的傳感器遙測。
• 構建和部署您的機器學習模型預測資產失敗之前發生。
• 創建一個實時數據庫使用磚三角洲的存儲和流傳感器部分和資產。

建立你的運動流

預測災難性的失敗,我們需要將資產從運動傳感器連續的數據流,流火花,我們流k - means模型。讓我們開始通過配置運動流使用下麵的代碼片段。潛水深度,請參考磚——亞馬遜運動的集成。

/ / = = =配置運動流= = =val awsAccessKeyId =“你的訪問密鑰ID”val awsSecretKey =“你的秘密密鑰”val kinesisStreamName =“你流的名字”val kinesisRegion =“你的地區”/ /如“us-west-2”進口com.amazonaws.services.kinesis.model.PutRecordRequest進口com.amazonaws.services.kinesis.AmazonKinesisClientBuilder進口com.amazonaws.auth。{DefaultAWSCredentialsProviderChain, BasicAWSCredentials}進口java.nio.ByteBuffer進口scala.util.Random

與你的憑證,您可以運行一個火花流查詢從動作讀單詞和用下麵的代碼片段。

/ /建立運動流val運動= spark.readStream.format (“運動”).option (“streamName”kinesisStreamName).option (“地區”kinesisRegion).option (“initialPosition”,“TRIM_HORIZON”).option (“awsAccessKey”awsAccessKeyId).option (“awsSecretKey”awsSecretKey).load ()/ /執行查詢DataFrame agaijnst運動流val結果= kinesis.selectExpr (“lcase (CAST(數據作為字符串))“).groupBy ($“單詞”).count ()/ /輸出顯示為一個條形圖顯示器(結果)

配置自己的運動流,寫這些話給你的運動流通過創建一個低級運動等客戶下麵的代碼片段,每5 s循環。

/ /創建AWS的低級運動客戶機Java SDK。val kinesisClient = AmazonKinesisClientBuilder.standard ().withRegion (kinesisRegion).withCredentials (新AWSStaticCredentialsProvider (新BasicAWSCredentials (awsAccessKeyId awsSecretKey))).build ()println (s“把話說到流中kinesisStreamName美元”)varlastSequenceNumber:字符串=零為(我
              
              

探索你的傳感器數據

之前,我們可以構建我們的模型來預測健康與損壞壓縮機,我們先來做一個小的數據探索。首先,我們需要導入我們的健康和損壞的壓縮機數據;下麵的代碼片段導入健康壓縮機以CSV格式的數據到一個SQL DataFrame火花。

//閱讀健康的壓縮機讀數(代表通過H1的前綴)val healthyCompressorsReadings=sqlContext.read.format (“com.databricks.spark.csv”). schema (StructType (StructField (“AN10”,倍增式,假)::StructField (“AN3”,倍增式,假)::StructField (“AN4”,倍增式,假)::StructField (“AN5”,倍增式,假)::StructField (“AN6”,倍增式,假)::StructField (“AN7”,倍增式,假)::StructField (“AN8”,倍增式,假)::StructField (“AN9”,倍增式,假)::StructField(“速度”,倍增式,假)::Nil)).load(“/壓縮機/ csv / H1 *”)//創建健康的壓縮機火花SQL表healthyCompressorsReadings.write.mode (SaveMode.Overwrite) .saveAsTable (“compressor_healthy”)//讀表從拚花val compressor_healthy=表(“compressor_healthy”)

我們還將數據保存為火花SQL表我們可以使用火花SQL查詢。例如,我們可以使用磚顯示命令來查看我們的表統計信息表損壞壓縮機。

顯示器(compressor_damaged.describe ())

在一個隨機樣本的健康和破壞數據使用下麵的代碼片段:

/ /獲得一個隨機樣本的健康和損壞壓縮機val randomSample = compressor_healthy.withColumn (“ReadingType”點燃(“健康”)采樣(假,500年/4800000.0).union (compressor_damaged.withColumn (“ReadingType”點燃(“受損”)采樣(假,500年/4800000.0))

我們可以使用磚顯示命令來可視化使用散點圖的隨機樣本數據。

/ /視圖散點圖的健康與損壞壓縮機讀數顯示器(randomSample)

建設我們的模型

接下來的步驟實現我們的預見性維護模式是創建一個k - means模型集群數據集來預測受損與健康的壓縮機。除了k - means成為一個受歡迎的和容易理解的聚類算法,也有使用流k - means模型的好處使我們輕鬆地執行相同的模型在批處理和流場景。

我們要做的第一件事是確定最優k值(即最優數量的集群)。我們目前確定的區別健康和損壞,憑直覺k的值是2讓我們驗證。正如下麵的代碼片段中,我們將構建一個毫升管道,所以我們可以很容易地重用我們的新數據集的模型(即上遊流數據集)。我們毫升管道相對簡單使用VectorAssembler來定義我們的特性包括空氣和噪音列(即列前的)和使用MinMaxScaler擴展它。

進口org.apache.spark.ml._進口org.apache.spark.ml.feature._進口org.apache.spark.ml.clustering._進口org.apache.spark.mllib.linalg.Vectors/ /使用KMeansModelval模型:數組[org.apache.spark.mllib.clustering。KMeansModel] =新數組(org.apache.spark.mllib.clustering.KMeansModel) (10)/ /使用VectorAssembler來定義我們的特性基於空氣+噪音列(和規模)val vectorAssembler =新VectorAssembler () .setInputCols (compressor_healthy.columns.filter (_.startsWith (“一個”))).setOutputCol (“特征”)val mmScaler =新MinMaxScaler () .setInputCol (“特征”).setOutputCol (“縮放”)/ /構建毫升管道val管道=新管道().setStages (數組(vectorAssembler mmScaler))/ /構建我們的模型基於健康壓縮機數據val prepModel = pipeline.fit (compressor_healthy)val prepData = prepModel.transform (compressor_healthy) .cache ()/ /遍曆找到最好的K值val麥克斯特=20.val maxK =5val findBestK =為(k
              
              我們運行一個數字的雖然迭代確定最佳的k值,為目的的這演示,我們< em > init </ em >自己k值(2…5]和set the max iterations to 20. The goal is to iterate through the various k and WSSSEem >(在集總和的的平方錯誤)值;最優k值(最理想的數字的集群)是一個有一個“肘部”在WSSSE圖。我們也可以計算出最高階導數的這個圖與下麵的代碼片段。
              
              

//計算導數的WSSSE val previousDf=kWssseDf。與Column("k", $"k"1)。withColumnRenamed (“wssse”、“previousWssse”) val derivativeOfWssse=previousDf。加入(kWssseDf“k”)。選擇Expr("k", "previousWssse - wssse derivative").orderBy($"k")//找到點與“最高”的導數//即最優數量的集群是bestK=2val bestK=derivativeOfWssse。選擇((鉛(“導數”,1)。在(Window.orderBy (“k”))- - - - - -美元“導數”)。作為(“nextDerivative”) (“k”) .orderBy美元“nextDerivative”。desc).rdd.map (_ (1).first.asInstanceOf [Int]

/ /創建“kmeans”模型val kmeans =新KMeans ().setK (bestK).setSeed (1L).setMaxIter (One hundred.).setFeaturesCol (“縮放”)val bestModel = kmeans.fit (prepData)val wssse = bestModel.computeCost (prepData)/ /輸出kmeans: org.apache.spark.ml.clustering。KMeans = kmeans_aeafe51274c3bestModel: org.apache.spark.ml.clustering。KMeansModel = kmeans_aeafe51274c3wssse:雙=329263.3539615829

/ /計算WSSSE損壞壓縮機val prepDamagedModel = pipeline.fit (compressor_damaged)val prepDamagedData = prepModel.transform (compressor_damaged) .cache ()val bestDamagedModel = kmeans.fit (prepDamagedData)val wssse = bestDamagedModel.computeCost (prepDamagedData)/ /輸出prepDamagedModel: org.apache.spark.ml。PipelineModel = pipeline_70af6bee9dadprepDamagedData: org.apache.spark.sql.Dataset [org.apache.spark.sql。行]= [AN10:雙AN3:雙…9多個字段)bestDamagedModel: org.apache.spark.ml.clustering。KMeansModel = kmeans_aeafe51274c3wssse:雙=1440111.9276810554

/ /創建StreamingKMeans()模型val kMeansModel =新StreamingKMeans ().setDecayFactor (0.5).setK (2).setRandomCenters (8,0.1)

/ /函數來創建一個新的StreamingContext和設置它def creatingFunc (): StreamingContext = {/ /創建一個StreamingContextval ssc =新StreamingContext (sc,秒(batchIntervalSeconds))val batchInterval =秒(batchIntervalSeconds)ssc.remember(秒(300年))
              val dstream = ssc.queueStream(隊列)/ / DStream接收數據,我們計算每個mini-batch WSSSE/ /保存這些數據DBFSkMeansModel.trainOn (dstream)dstream。為eachRDD {抽樣= >val wssse = kMeansModel.latestModel () .computeCost(抽樣)val timestamp =係統。currentTimeMillis /1000年sc.parallelize (Seq (WsseTimestamps(時間戳,wssse))).toDF () .write.mode (SaveMode.Append) . json (“/ tmp /壓縮機”)}println (“創建函數創建新的StreamingContext呼籲壓縮機故障預測”)newContextCreated =真正的ssc}

/ /執行火花流上下文val ssc = StreamingContext.getActiveOrCreate (creatingFunc)如果(newContextCreated) {println (“從當前定義創建新的上下文創建函數”)}其他的{println (“現有上下文運行或從檢查站中恢複過來,可能不是運行當前定義創建函數”)}/ /開始引發流上下文ssc.start ()

/ /讀取StreamingKMeans DBFS()結果val compressorsResults = sqlContext.read.json (“/ tmp /壓縮機”)/ /視圖模型顯示器(compressorsResults.orderBy (“t”))

//創建健康的壓縮機磚δ表創建表compressor_healthy (AN10雙,AN3雙,AN4雙,AN5雙,AN6雙,AN7雙,AN8雙,AN9雙,速度雙)使用δ選項(路徑“/壓縮機/δ/健康/”)