Auf einen Blick: Die Verwendung von Apache Spark für die prädiktive Wartung nutzt maschinelles Lernen, um Maschinen zu identifizieren, die Service benötigen. Der Leitfaden erklärt, wie man Spark-Jobs für Vorhersagen einreicht, mit wichtigen Codeschnippeln und Anweisungen.
- Reichen Sie Spark-Aufträge für Wartungsprognosen ein.
- Nutzen Sie maschinelles Lernen, um Maschinen zu identifizieren, die Wartung benötigen.
Im ersten Teil haben wir erklärt, wie man ein Trainingsmodell erstellt. In diesem Teil zeigen wir, wie wir Vorhersagen treffen können, um zu zeigen, welche Maschinen in unserem Datensatz für Wartungsarbeiten außer Betrieb genommen werden sollten.
Zuerst erfährst du, wie du den Auftrag mit spark-submit bei Spark einreichst:
- JAR-Datei enthält, die com.bmc.lr.makePrediction enthält
- Welche Datei man lesen sollte, also die, die du gerade in Teil I generiert hast (Link zum vorherigen Artikel setzen)
- welches Modell soll verwendet werden, also das, das du gerade in Teil I generiert hast (Link zum vorherigen Artikel hinzufügen)
spark-submit
--klasse com.bmc.lr.makePrediction
--Meister lokal[*]
hdfs://localhost:9000/maintenance/lr-assembly-1.0.jar
hdfs://localhost:9000/maintenance/2018.04.20.15.48.54.csv
hdfs://localhost:9000/maintenance/maintenance_model
spark-submit --class com.bmc.lr.makePrediction --master local[*] hdfs://localhost:9000/maintenance/lr-assembly-1.0.jar hdfs://localhost:9000/maintenance/2018.04.20.15.48.54.csv hdfs://localhost:9000/maintenance/maintenance_model
Code
Hier sind die üblichen Importe und der Paketname.
package com.bmc.lr
import org.apache.spark.sql.SQLContext
import org.apache.spark.ml.classification.{BinaryLogisticRegressionSummary, LogisticRegression}
import org.apache.spark.mllib.linalg.{Vector, Vectors}
import org.apache.spark.ml.evaluation.BinaryClassificationEvaluator
import org.apache.spark.ml.feature.{VectorAssembler, StringIndexer}
import org.apache.spark.ml.classification.LogisticRegressionModel
import org.apache.spark.SparkConf
import org.apache.spark.sql.SQLContext
import org.apache.spark.SparkContext
Lesen Sie die Kommandozeilenargumente und richten Sie den Spark- und SQL-Kontext ein. Erstellen Sie das Objekt makePrediction, damit wir die klasse com.bmc.lr.makePrediction aus spark-submit instanziieren können.
object makePrediction {
def main(args: Array[String]): Unit = {
val conf = new SparkConf().setAppName("lr")
val sc = new SparkContext(conf)
var modelFile = args(1);
var file = args(0);
val sqlContext = new SQLContext(sc)
Verwenden Sie Databricks, um die Eingabedatei zu lesen und erstellen Sie einen org.apache.spark.sql.DataFrame. Spätere Databricks sind nützlich, um die Ausgabe in Hadoop zu speichern, da sie nur eine Codezeile benötigen.
val df = sqlContext.read.format("com.databricks.spark.csv").option("header", "true").option("inferSchema", "true").option("delimiter",";").load(file)
val df = sqlContext.read.format("com.databricks.spark.csv").option("header", "true").option("inferSchema", "true").option("delimiter",";").load(file)
An diesem Punkt sieht df so aus:
lifetime|broken| pressureInd| moistureInd| temperatureInd|team| provider|
+------------------+------+------------------+------------------+--------------
|17.792044609594086| 0|104.25543343273085| 88.67373165099184|122.75111030066334| 47|Ford F-750|
| 69.39805687060903| 0| 93.34561942201603| 86.62996015487022| 84.9796059428202| 34|Ford F-750|
| 84.53664924532875| 0|110.64579687466193|125.89351825805036| 58.34915688191312| 8|Ford F-750|
org.apache.spark.ml.featureVectorAssembler transformiert die Features in featureCols in eine Vektorspalte. Wir benötigen es in diesem Format, um es in LogisticRegressionModel.transform() einzubinden, das die Features (lifetime, pressureInd, moistureInd, temperatureInd) und Labels (gebrochen , die wir zur besseren Verständlichkeit umbenennen) als Schritt nimmt, um vorherzusagen, welche Maschinen ausfallen werden.
Dann nehmen wir das gebrochene (1 oder 0) Label und fügen es in org.apache.spark.ml.feature.StringIndexer ein, um einen einzelnen Wert zu enthalten, das Label. org.apache.spark.ml.feature.StringIndexer.fit() passt die Daten an das Modell an, also die Vorhersagen treffen.
You can see what the features vectors look like after we create df3 (see below).
val featureCols = Array("lifetime", "pressureInd", "moistureInd", "temperatureInd")
val assembler = new VectorAssembler().setInputCols(featureCols).setOutputCol("features")
val labelIndexer = new StringIndexer().setInputCol("broken").setOutputCol("label")
val df2 = assembler.transform(df)
val df3 = labelIndexer.fit(df2).transform(df2)
Hier zeigen wir die Spalte Merkmale (wenn auch in der Darstellung abgeschnitten). Wenn Sie sich erinnern, ist das wie die dichten Vektoren, die wir verwenden, um sie in Trainingsmodelle einzubinden.
df3.select("features").show()
+--------------------+
| features|
+--------------------+
|[17.7920446095940...|
|[69.3980568706090...|
|[84.5366492453287...|
|[71.6002965259175...|
|[46.8176995523900...|
Lade das gespeicherte org.apache.spark.ml.classification.LogisticRegressionModel Modell aus dem Hadoop-Dateisystem. Wir haben dieses Modell im ersten Programm trainiert.
val model = LogisticRegressionModel.load(modelFile)
val model = LogisticRegressionModel.load(modelFile)
Hier wählen wir nur die Spalten aus, die wir wollen (wobei wir Features entfernen, da es schwer zu drucken und anzuzeigen und auch redundant ist). Dann zeigen wir nur die Maschinen, die Wartung benötigen, also solche, deren vorhergesagter Wert von Broken 1 ist (wahr).
val predictions = model.transform(df3)
var df4 = predictions.select ("team", "provider", "pressureInd", "moistureInd", "temperatureInd", "label", "prediction")
val df5 = df4.filter("prediction=1")
df5.show()
Bei df5.show() sehen wir, dass 4 Maschinen gewartet werden müssen:
+----+----------+------------------+------------------+-----------------+-----+----------+
|team| provider| pressureInd| moistureInd| temperatureInd|label|prediction|
+----+----------+------------------+------------------+-----------------+-----+----------+
| 34|Ford F-750| 93.34561942201603| 86.62996015487022| 84.9796059428202| 0.0| 1.0|
| 8|Ford F-750|110.64579687466193|125.89351825805036|58.34915688191312| 0.0| 1.0|
| 83|Ford F-750| 55.77009802003853| 66.832777175712|125.2982705340028| 0.0| 1.0|
| 2|Ford F-750| 84.1763960666348| 82.1342684415311|57.73202884026434| 0.0| 1.0|
+----+----------+------------------+------------------+-----------------+-----+----------+
Jetzt sehen wir ein weiteres Beispiel dafür, wie nützlich Databricks sind, indem wir die Vorhersagen in einem .csv in Hadoop schreiben können. Wir speichern die Ausgabe in einer Datei, deren Name das Datum ist. Beachten Sie auch, dass wir java.util.Date importieren, da dies der einfachste Weg ist, verschiedene Datumsformate zu erstellen.
import java.util.Date
import java.text.SimpleDateFormat
val date = new Date()
var dformat:SimpleDateFormat = new SimpleDateFormat("yyyy.MM.dd.HH.mm.ss");
val csvFile = "hdfs://localhost:9000/maintenance/" + dformat.format(date) + ".csv"
df5.write.format("com.databricks.spark.csv").option("header", "true").save(csvFile)
}
}
Jetzt zeigen wir die Daten:
hadoop fs -cat hdfs://localhost:9000/maintenance/2018.05.16.09.07.33.csv/part-00000-7a9a4de6-ab90-4cd4-b9f5-d0d30c881b2c-c000.csv
Team, Anbieter, Druck, Feuchtigkeit, Temperatur, Etikett, Vorhersage
34,Ford F-750,93.34561942201603,86,62996015487022,84.9796059428202,0.0,1.0
8,Ford F-750,110.64579687466193.125.89351825805036,58.34915688191312,0.0,1.0
83,Ford F-750,55.77009802003853,66.832777175712.125.2982705340028,0.0,1.0
2,Ford F-750,84.1763960666348,82.1342684415311,57.73202884026434,0.0,1.0
hadoop fs -cat hdfs://localhost:9000/maintenance/2018.05.16.09.07.33.csv/part-00000-7a9a4de6-ab90-4cd4-b9f5-d0d30c881b2c-c000.csv team,provider,pressureInd,moistureInd,temperatureInd,label,prediction 34,Ford F-750,93.34561942201603,86.62996015487022,84.9796059428202,0.0,1.0 8,Ford F-750,110.64579687466193,125.89351825805036,58.34915688191312,0.0,1.0 83,Ford F-750,55.77009802003853,66.832777175712,125.2982705340028,0.0,1.0 2,Ford F-750,84.1763960666348,82.1342684415311,57.73202884026434,0.0,1.0
Apache Spark Leitfaden
- Apache Sparks maschinelle Lernpipeline: Eine Einführung
- SGD-Beispiel für lineare Regression mit Apache Spark
- Spark Decision Tree Klassifikator
- Mit logistischer Regression, Scala und Spark
- Wie man Jupyter-Notizbücher mit Apache Spark verwendet
- Verwendung von Python und Spark Machine Learning zur Klassifikation
- Wie man Spark UDFs (benutzerdefinierte Funktionen) in Python schreibt
- Python Spark ML K-Means Beispiel
- Wie man maschinelles Lernen auf die Cybersicherheit anwendet
- Apache Spark: Zusammenarbeit mit Streams
- K-bedeutet Clustering mit Apache Spark
- Spark mit Hive verwenden
- Wie man Apache Spark verwendet, um Vorhersagen für die vorbeugende Wartung zu treffen.
Diese Beiträge sind meine eigenen und spiegeln nicht unbedingt die Position, Strategien oder Meinung von BMC wider.