让你更聪明的石油和天然气资产实施预见性维护砖

如何构建一个端到端的预测和砖三角洲和火花的流媒体数据管道吗
在砖试试这个笔记本

维护资产,如压缩机是一个极其复杂的努力:它们用于从小型钻机到深水平台、资产遍布全球各地,他们每天生成tb的数据。bob体育客户端下载失败只是一个压缩机成本每天数百万美元的生产损失。节省时间和金钱的一个重要方法是使用机器学习来预测故障维修工作指令和问题之前发生故障。

最后,你需要建立一个端到端的预测数据管道,可以提供一个实时数据库保持资产部分和传感器映射,支持连续大量的遥测应用程序流程,并允许您对这些数据集预测压缩机故障。

我们解决这些问题的方法是通过选择一个统一的平台,提供了这些功能。bob体育客户端下载砖提供了一个bob体育亚洲版统一的分析平台bob体育客户端下载汇集了大数据和人工智能的不同角色,并允许您的组织一起合作在一个工作区。砖的其他重要的优势统一分析平台包括的能力:bob体育亚洲版bob体育客户端下载

• 自旋向上必要的资源和数据科学家、工程师、数据和数据分析师快速理解数据。
• 有多重云策略允许每个人都使用相同的协作工作区Azure或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双,速度双)使用δ选项(路径“/压缩机/δ/健康/”)