Erstellen Sie mit Amazon Kinesis, AWS Glue und Amazon SageMaker eine vorausschauende Wartungslösung

Unternehmen bauen und verwenden zunehmend auf maschinellem Lernen (ML) basierende Lösungen für eine Vielzahl von Anwendungsfällen und Problemen, darunter vorausschauende Wartung von Maschinenteilen, Produktempfehlungen auf der Grundlage von Kundenpräferenzen, Kreditprofilerstellung, Inhaltsmoderation, Betrugserkennung und mehr. In vielen dieser Szenarien können die Effektivität und die Vorteile dieser ML-gestützten Lösungen weiter verbessert werden, wenn sie Datenereignisse nahezu in Echtzeit verarbeiten und daraus Erkenntnisse ableiten können.

Obwohl der geschäftliche Wert und die Vorteile von ML-gestützten Lösungen in nahezu Echtzeit bekannt sind, ist die Architektur, die erforderlich ist, um diese Lösungen in großem Maßstab mit optimaler Zuverlässigkeit und Leistung zu implementieren, kompliziert. Dieser Beitrag beschreibt, wie Sie kombinieren können Amazon Kinesis, AWS-Kleber und Amazon Sage Maker um eine Feature-Engineering- und Inferenzlösung für Predictive Maintenance nahezu in Echtzeit zu entwickeln.

Anwendungsfallübersicht

Wir konzentrieren uns auf einen Anwendungsfall der vorausschauenden Wartung, bei dem im Feld eingesetzte Sensoren (z. B. Industrieanlagen oder Netzwerkgeräte) ersetzt oder repariert werden müssen, bevor sie fehlerhaft werden und Ausfallzeiten verursachen. Ausfallzeiten können für Unternehmen teuer werden und zu einem schlechten Kundenerlebnis führen. Die durch ein ML-Modell unterstützte vorausschauende Wartung kann auch dazu beitragen, die regelmäßigen planbasierten Wartungszyklen zu erweitern, indem sie informiert, wenn ein Maschinenteil in gutem Zustand nicht ersetzt werden sollte, wodurch unnötige Kosten vermieden werden.

In diesem Beitrag konzentrieren wir uns auf die Anwendung von maschinellem Lernen auf einen synthetischen Datensatz, der Maschinenausfälle aufgrund von Merkmalen wie Lufttemperatur, Prozesstemperatur, Drehzahl, Drehmoment und Werkzeugverschleiß enthält. Der verwendete Datensatz stammt aus der UCI-Datenspeicher.

Maschinenversagen besteht aus fünf unabhängigen Fehlermodi:

• Werkzeugverschleißfehler (TWF)
• Wärmeableitungsfehler (HDF)
• Stromausfall (PWF)
• Überlastungsfehler (OSF)
• Zufälliger Fehler (RNF)

Das Maschinenfehleretikett gibt an, ob die Maschine für einen bestimmten Datenpunkt ausgefallen ist, wenn einer der vorhergehenden Fehlermodi zutrifft. Wenn mindestens einer der Fehlermodi wahr ist, schlägt der Prozess fehl und das Maschinenfehlerlabel wird auf 1 gesetzt. Das Ziel des ML-Modells ist es, Maschinenfehler korrekt zu identifizieren, damit eine nachgelagerte vorausschauende Wartungsmaßnahme eingeleitet werden kann.

Lösungsüberblick

Für unseren Anwendungsfall der vorausschauenden Wartung gehen wir davon aus, dass Gerätesensoren verschiedene Messungen und Messwerte über Maschinenteile streamen. Unsere Lösung nimmt dann jedes Mal einen Teil der Streaming-Daten (Mikro-Batch) und führt die Verarbeitung und das Feature-Engineering durch, um Features zu erstellen. Die erstellten Funktionen werden dann verwendet, um Inferenzen aus einem trainierten und bereitgestellten ML-Modell nahezu in Echtzeit zu generieren. Die generierten Inferenzen können von nachgelagerten Anwendungen weiter verarbeitet und genutzt werden, um geeignete Maßnahmen zu ergreifen und Wartungsaktivitäten einzuleiten.

Das folgende Diagramm zeigt die Architektur unserer Gesamtlösung.

Die Lösung besteht im Großen und Ganzen aus den folgenden Abschnitten, die später in diesem Beitrag ausführlich erläutert werden:

• Streaming-Datenquelle und Aufnahme - Wir gebrauchen Amazon Kinesis-Datenströme um Streaming-Daten von den Feldsensoren in großem Maßstab zu sammeln und für die weitere Verarbeitung bereitzustellen.
• Feature-Engineering nahezu in Echtzeit – Wir verwenden AWS Glue-Streaming-Jobs, um Daten aus einem Kinesis-Datenstrom zu lesen und Datenverarbeitung und Feature-Engineering durchzuführen, bevor wir die abgeleiteten Features darin speichern Amazon Simple Storage-Service (Amazon S3). Amazon S3 bietet eine zuverlässige und kostengünstige Option zum Speichern großer Datenmengen.
• Modellschulung und -einsatz – Wir verwenden den AI4I Predictive Maintenance-Datensatz aus dem UCI Data Repository, um ein ML-Modell basierend auf dem XGBoost-Algorithmus mit SageMaker zu trainieren. Anschließend stellen wir das trainierte Modell auf einem SageMaker-Endpunkt für asynchrone Inferenzen bereit.
• ML-Inferenz nahezu in Echtzeit – Nachdem die Funktionen in Amazon S3 verfügbar sind, müssen wir Inferenzen aus dem bereitgestellten Modell nahezu in Echtzeit generieren. SageMaker-Endpunkte für asynchrone Inferenzen eignen sich gut für diese Anforderung, da sie größere Payload-Größen (bis zu 1 GB) unterstützen und Inferenzen innerhalb von Minuten (bis zu maximal 15 Minuten) generieren können. Wir verwenden S3-Ereignisbenachrichtigungen, um eine auszuführen AWS Lambda -Funktion zum Aufrufen eines asynchronen Inferenzendpunkts von SageMaker. SageMaker-Endpunkte für asynchrone Inferenz akzeptieren S3-Standorte als Eingabe, generieren Inferenzen aus dem bereitgestellten Modell und schreiben diese Inferenzen nahezu in Echtzeit an Amazon S3 zurück.

Der Quellcode für diese Lösung befindet sich auf GitHub. Die Lösung wurde getestet und sollte in us-east-1 ausgeführt werden.

Wir verwenden eine AWS CloudFormation Vorlage, bereitgestellt mit AWS Serverless-Anwendungsmodell (AWS SAM) und SageMaker-Notebooks zum Bereitstellen der Lösung.

Voraussetzungen:

Um loszulegen, müssen Sie als Voraussetzung die haben SAM-CLI, Python 3 und PIP Eingerichtet. Die muss man auch haben AWS-Befehlszeilenschnittstelle (AWS CLI) richtig konfiguriert.

Stellen Sie die Lösung bereit

Sie können verwenden AWS CloudShell um diese Schritte auszuführen. CloudShell ist eine browserbasierte Shell, die mit Ihren Konsolenanmeldeinformationen vorauthentifiziert wird und vorinstallierte allgemeine Entwicklungs- und Betriebstools (wie AWS SAM, AWS CLI und Python) enthält. Daher ist keine lokale Installation oder Konfiguration erforderlich.

• Wir beginnen mit der Erstellung eines S3-Buckets, in dem wir das Skript für unseren AWS Glue-Streaming-Job speichern. Führen Sie den folgenden Befehl in Ihrem Terminal aus, um einen neuen Bucket zu erstellen:

aws s3api create-bucket --bucket sample-script-bucket-$RANDOM --region us-east-1

• Notieren Sie sich den Namen des erstellten Buckets.

• Als Nächstes klonen wir lokal das Code-Repository, das die CloudFormation-Vorlage zum Bereitstellen des Stacks enthält. Führen Sie den folgenden Befehl in Ihrem Terminal aus:

git clone https://github.com/aws-samples/amazon-sagemaker-predictive-maintenance

• Navigieren Sie zum sam-template-Verzeichnis:

cd amazon-sagemaker-predictive-maintenance/sam-template

• Führen Sie den folgenden Befehl aus, um das AWS Glue-Auftragsskript (aus glue_streaming/app.py) in den von Ihnen erstellten S3-Bucket zu kopieren:

aws s3 cp glue_streaming/app.py s3://sample-script-bucket-30232/glue_streaming/app.py

• Sie können jetzt mit dem Erstellen und Bereitstellen der Lösung über die CloudFormation-Vorlage über AWS SAM fortfahren. Führen Sie den folgenden Befehl aus:

sam deploy --guided

• Geben Sie Argumente für die Bereitstellung an, z. B. den Stack-Namen, die bevorzugte AWS-Region (us-east-1), Und GlueScriptsBucket.

Stellen Sie sicher, dass Sie denselben S3-Bucket bereitstellen, den Sie zuvor für den AWS Glue-Skript-S3-Bucket erstellt haben (parameter GlueScriptsBucket im folgenden Screenshot).

Nachdem Sie die erforderlichen Argumente angegeben haben, startet AWS SAM die Stack-Bereitstellung. Der folgende Screenshot zeigt die erstellten Ressourcen.

Nachdem der Stack erfolgreich bereitgestellt wurde, sollte die folgende Meldung angezeigt werden.

• Öffnen Sie in der AWS CloudFormation-Konsole den Stack (für diesen Beitrag nrt-streaming-inference), die beim Bereitstellen der CloudFormation-Vorlage bereitgestellt wurde.
• Auf dem Downloads Notieren Sie sich auf der Registerkarte SageMaker die Instanz-ID des Notebooks.

• Öffnen Sie diese Instanz in der SageMaker-Konsole.

Die SageMaker-Notebookinstanz hat bereits die erforderlichen Notebooks vorinstalliert.

Navigieren Sie zum Notizbuchordner und öffnen und befolgen Sie die Anweisungen in den Notizbüchern (Data_Pre-Processing.ipynb und ModelTraining-Evaluation-and-Deployment.ipynb), um das Dataset zu untersuchen, Vorverarbeitung und Feature-Engineering durchzuführen und das Modell zu trainieren und an einem SageMaker-Endpunkt für asynchrone Inferenz bereitzustellen.

Streaming-Datenquelle und Aufnahme

Kinesis Data Streams ist ein serverloser, skalierbarer und dauerhafter Echtzeit-Daten-Streaming-Dienst, mit dem Sie große Ströme von Datensätzen in Echtzeit erfassen und verarbeiten können. Kinesis Data Streams ermöglicht das Erfassen, Verarbeiten und Speichern von Datenströmen aus einer Vielzahl von Quellen, z. B. Protokolldaten der IT-Infrastruktur, Anwendungsprotokolle, soziale Medien, Marktdaten-Feeds, Web-Clickstream-Daten, IoT-Geräte und -Sensoren und mehr. Sie können einen Kinesis-Datenstrom je nach Durchsatz- und Skalierungsanforderungen im On-Demand-Modus oder im bereitgestellten Modus bereitstellen. Weitere Informationen finden Sie unter Auswählen des Datenstrom-Kapazitätsmodus.

Für unseren Anwendungsfall gehen wir davon aus, dass verschiedene Sensoren Messwerte wie Temperatur, Drehzahl, Drehmoment und Werkzeugverschleiß an einen Datenstrom senden. Kinesis Data Streams fungiert als Trichter zum Sammeln und Aufnehmen von Datenströmen.

Wir nutzen die Amazon Kinesis-Datengenerator (KDG) später in diesem Beitrag, um Daten zu generieren und an einen Kinesis-Datenstrom zu senden, wodurch Daten simuliert werden, die von Sensoren generiert werden. Die Daten aus dem Datenstrom sensor-data-stream werden mithilfe eines AWS Glue-Streaming-Jobs aufgenommen und verarbeitet, den wir als Nächstes besprechen.

Feature-Engineering nahezu in Echtzeit

AWS Glue-Streaming-Jobs bieten eine bequeme Möglichkeit, Streaming-Daten in großem Umfang zu verarbeiten, ohne die Rechenumgebung verwalten zu müssen. AWS Glue ermöglicht Ihnen das Ausführen von Extrahierungs-, Transformations- und Ladevorgängen (ETL) für Streaming-Daten mithilfe von kontinuierlich ausgeführten Jobs. AWS Glue-Streaming-ETL basiert auf der Apache Spark Structured Streaming-Engine und kann Streams von Kinesis, Apache Kafka und Amazon Managed Streaming für Apache Kafka (Amazon MSK).

Der Streaming-ETL-Auftrag kann sowohl integrierte AWS Glue-Transformationen als auch Transformationen verwenden, die in Apache Spark Structured Streaming nativ sind. Sie können auch Spark ML und verwenden MLLib Bibliotheken in AWS Glue-Jobs für eine einfachere Feature-Verarbeitung mit leicht verfügbaren Hilfsbibliotheken.

Wenn das Schema der Streaming-Datenquelle vorbestimmt ist, können Sie es in einer AWS Data Catalog-Tabelle angeben. Wenn die Schemadefinition nicht vorher bestimmt werden kann, können Sie die Schemaerkennung im Streaming-ETL-Job aktivieren. Der Job ermittelt dann automatisch das Schema aus den eingehenden Daten. Zusätzlich können Sie die verwenden AWS Glue-Schema-Registrierung um eine zentrale Erkennung, Kontrolle und Weiterentwicklung von Datenstromschemata zu ermöglichen. Sie können die Schemaregistrierung weiter in den Datenkatalog integrieren, um optional in der Schemaregistrierung gespeicherte Schemas zu verwenden, wenn Sie AWS Glue-Tabellen oder -Partitionen im Datenkatalog erstellen oder aktualisieren.

Für diesen Beitrag erstellen wir eine AWS Glue Data Catalog-Tabelle (sensor-stream) mit unserem Kinesis-Datenstrom als Quelle und definieren das Schema für unsere Sensordaten.

Wir erstellen einen dynamischen AWS Glue-Datenrahmen aus der Data Catalog-Tabelle, um die Streaming-Daten von Kinesis zu lesen. Wir geben auch die folgenden Optionen an:

• Eine Fenstergröße von 60 Sekunden, damit der AWS Glue-Job Daten in 60-Sekunden-Fenstern liest und verarbeitet
• Die Ausgangsposition TRIM_HORIZON, um das Lesen aus den ältesten Datensätzen im Kinesis-Datenstrom zu ermöglichen

Wir verwenden auch Spark MLlibs StringIndexer Feature Transformer zum Codieren des String-Spaltentyps in Label-Indizes. Diese Transformation wird mithilfe von Spark ML-Pipelines implementiert. Spark-ML-Pipelines Bereitstellung eines einheitlichen Satzes von High-Level-APIs für ML-Algorithmen, um die Kombination mehrerer Algorithmen in einer einzigen Pipeline oder einem einzigen Workflow zu vereinfachen.

Wir verwenden die foreachBatch-API, um eine Funktion namens processBatch aufzurufen, die wiederum die von diesem Datenrahmen referenzierten Daten verarbeitet. Siehe folgenden Code:

# Read from Kinesis Data Stream
sourceStreamData = glueContext.create_data_frame.from_catalog(database = "sensordb", table_name = "sensor-stream", transformation_ctx = "sourceStreamData", additional_options = {"startingPosition": "TRIM_HORIZON"})
type_indexer = StringIndexer(inputCol="type", outputCol="type_enc", stringOrderType="alphabetAsc")
pipeline = Pipeline(stages=[type_indexer])
glueContext.forEachBatch(frame = sourceStreamData, batch_function = processBatch, options = {"windowSize": "60 seconds", "checkpointLocation": checkpoint_location})

Die Funktion processBatch führt die angegebenen Transformationen durch und partitioniert die Daten in Amazon S3 basierend auf Jahr, Monat, Tag und Batch-ID.

Wir partitionieren auch die AWS Glue-Partitionen in eine einzige Partition neu, um zu vermeiden, dass zu viele kleine Dateien in Amazon S3 vorhanden sind. Das Vorhandensein mehrerer kleiner Dateien kann die Leseleistung beeinträchtigen, da es den Aufwand für das Suchen, Öffnen und Lesen jeder Datei erhöht. Schließlich schreiben wir die Features zum Generieren von Inferenzen in ein Präfix (Features) innerhalb des S3-Buckets. Siehe folgenden Code:

# Function that gets called to perform processing, feature engineering and writes to S3 for every micro batch of streaming data from Kinesis.
def processBatch(data_frame, batchId):
transformer = pipeline.fit(data_frame)
now = datetime.datetime.now()
year = now.year
month = now.month
day = now.day
hour = now.hour
minute = now.minute
if (data_frame.count() > 0):
data_frame = transformer.transform(data_frame)
data_frame = data_frame.drop("type")
data_frame = DynamicFrame.fromDF(data_frame, glueContext, "from_data_frame")
data_frame.printSchema()
# Write output features to S3
s3prefix = "features" + "/year=" + "{:0>4}".format(str(year)) + "/month=" + "{:0>2}".format(str(month)) + "/day=" + "{:0>2}".format(str(day)) + "/hour=" + "{:0>2}".format(str(hour)) + "/min=" + "{:0>2}".format(str(minute)) + "/batchid=" + str(batchId)
s3path = "s3://" + out_bucket_name + "/" + s3prefix + "/"
print("-------write start time------------")
print(str(datetime.datetime.now()))
data_frame = data_frame.toDF().repartition(1)
data_frame.write.mode("overwrite").option("header",False).csv(s3path)
print("-------write end time------------")
print(str(datetime.datetime.now()))

Modellschulung und -bereitstellung

SageMaker ist ein vollständig verwalteter und integrierter ML-Dienst, der es Data Scientists und ML-Ingenieuren ermöglicht, ML-Modelle schnell und einfach zu erstellen, zu trainieren und bereitzustellen.

Innerhalb des Data_Pre-Processing.ipynb-Notebooks, Wir importieren zunächst den AI4I Predictive Maintenance-Datensatz aus dem UCI Data Repository und führen eine explorative Datenanalyse (EDA) durch. Wir führen auch Feature-Engineering durch, um unsere Features für das Training des Modells nützlicher zu machen.

Zum Beispiel haben wir innerhalb des Datensatzes ein Merkmal namens Typ, das den Qualitätstyp des Produkts als L (niedrig), M (mittel) oder H (hoch) darstellt. Da es sich um eine kategoriale Funktion handelt, müssen wir sie codieren, bevor wir unser Modell trainieren. Wir verwenden den LabelEncoder von Scikit-Learn, um dies zu erreichen:

from sklearn.preprocessing import LabelEncoder
type_encoder = LabelEncoder()
type_encoder.fit(origdf['type'])
type_values = type_encoder.transform(origdf['type'])

Nachdem die Features verarbeitet und die kuratierten Trainings- und Testdatensätze generiert wurden, sind wir bereit, ein ML-Modell zu trainieren, um basierend auf Systemmesswerten vorherzusagen, ob die Maschine ausgefallen ist oder nicht. Wir trainieren ein XGBoost-Modell mit dem integrierten Algorithmus von SageMaker. XGBoost kann gute Ergebnisse für mehrere Arten von ML-Problemen liefern, einschließlich der Klassifizierung, selbst wenn die Trainingsbeispiele begrenzt sind.

SageMaker-Schulungsjobs bieten eine leistungsstarke und flexible Möglichkeit, ML-Modelle auf SageMaker zu trainieren. SageMaker verwaltet die zugrunde liegende Recheninfrastruktur und stellt bereit mehrere Optionen je nach Anwendungsfall für unterschiedliche Modellschulungsanforderungen zur Auswahl.

xgb = sagemaker.estimator.Estimator(container,
role,
instance_count=1,
instance_type='ml.c4.4xlarge',
output_path=xgb_upload_location,
sagemaker_session=sagemaker_session)
xgb.set_hyperparameters(max_depth=5,
eta=0.2,
gamma=4,
min_child_weight=6,
subsample=0.8,
silent=0,
objective='binary:hinge',
num_round=100)

xgb.fit({'train': s3_train_channel, 'validation': s3_valid_channel})

Wenn das Modelltraining abgeschlossen ist und die Modellbewertung basierend auf den Geschäftsanforderungen zufriedenstellend ist, können wir mit der Modellbereitstellung beginnen. Wir erstellen zunächst eine Endpunktkonfiguration mit der Objektoption AsyncInferenceConfig und verwenden das zuvor trainierte Modell:

endpoint_config_name = resource_name.format("EndpointConfig")
create_endpoint_config_response = sm_client.create_endpoint_config(
EndpointConfigName=endpoint_config_name,
ProductionVariants=[
{
"VariantName": "variant1",
"ModelName": model_name,
"InstanceType": "ml.m5.xlarge",
"InitialInstanceCount": 1,
}
],
AsyncInferenceConfig={
"OutputConfig": {
"S3OutputPath": f"s3://{bucket}/{prefix}/output",
#Specify Amazon SNS topics
"NotificationConfig": {
"SuccessTopic": "arn:aws:sns:<region>:<account-id>:<success-sns-topic>",
"ErrorTopic": "arn:aws:sns:<region>:<account-id>:<error-sns-topic>",
}},
"ClientConfig": {"MaxConcurrentInvocationsPerInstance": 4},
},)

Anschließend erstellen wir unter Verwendung der von uns erstellten Endpunktkonfiguration einen SageMaker-Endpunkt für asynchrone Inferenzen. Nach der Bereitstellung können wir damit beginnen, den Endpunkt aufzurufen, um asynchron Inferenzen zu generieren.

endpoint_name = resource_name.format("Endpoint")
create_endpoint_response = sm_client.create_endpoint(
EndpointName=endpoint_name, EndpointConfigName=endpoint_config_name)

Inferenz nahezu in Echtzeit

SageMaker asynchrone Inferenz Endpunkte bieten die Möglichkeit, eingehende Inferenzanforderungen in eine Warteschlange einzureihen und sie asynchron nahezu in Echtzeit zu verarbeiten. Dies ist ideal für Anwendungen, die Inferenzanforderungen mit größeren Nutzlastgrößen (bis zu 1 GB) haben, möglicherweise längere Verarbeitungszeiten (bis zu 15 Minuten) erfordern und Latenzanforderungen nahezu in Echtzeit haben. Durch die asynchrone Inferenz können Sie auch Kosten sparen, indem Sie die Anzahl der Instanzen automatisch auf Null skalieren, wenn keine Anfragen zu verarbeiten sind, sodass Sie nur bezahlen, wenn Ihr Endpunkt Anfragen verarbeitet.

Sie können einen asynchronen SageMaker-Inferenzendpunkt ähnlich wie einen Echtzeit-Inferenzendpunkt erstellen und zusätzlich die AsyncInferenceConfig -Objekt, während Sie Ihre Endpunktkonfiguration mit dem EndpointConfig-Feld in der CreateEndpointConfig-API erstellen. Das folgende Diagramm zeigt den Inferenzworkflow und wie ein asynchroner Inferenzendpunkt eine Inferenz generiert.

Um den asynchronen Inferenzendpunkt aufzurufen, sollte die Nutzlast der Anfrage in Amazon S3 gespeichert werden und ein Verweis auf diese Nutzlast muss als Teil der InvokeEndpointAsync-Anfrage bereitgestellt werden. Beim Aufruf stellt SageMaker die Anforderung zur Verarbeitung in die Warteschlange und gibt als Antwort eine Kennung und einen Ausgabeort zurück. Nach der Verarbeitung legt SageMaker das Ergebnis im Amazon S3-Speicherort ab. Sie können optional auswählen, Erfolgs- oder Fehlerbenachrichtigungen mit zu erhalten Amazon Simple Notification Service (Amazon SNS).

Testen Sie die End-to-End-Lösung

Führen Sie die folgenden Schritte aus, um die Lösung zu testen:

• Öffnen Sie in der AWS CloudFormation-Konsole den Stack, den Sie zuvor erstellt haben (nrt-streaming-inference).
• Auf dem Ausgänge kopieren Sie den Namen des S3-Buckets (EventsBucket).

Dies ist der S3-Bucket, in den unser AWS Glue-Streaming-Job Features schreibt, nachdem er den Kinesis-Datenstrom gelesen und verarbeitet hat.

Als Nächstes richten wir Ereignisbenachrichtigungen für diesen S3-Bucket ein.

• Navigieren Sie auf der Amazon S3-Konsole zum Bucket EventsBucket.
• Auf dem Ferienhäuser Registerkarte, in der Ereignisbenachrichtigungen Wählen Sie im Abschnitt Ereignisbenachrichtigung erstellen.

• Aussichten für Veranstaltungsname, eingeben invoke-endpoint-lambda.
• Aussichten für Vorsilbe, eingeben features/.
• Aussichten für Suffix, eingeben .csv.
• Aussichten für EreignistypenWählen Alle Objekterstellungsereignisse.

• Aussichten für ReisezielWählen Lambda-Funktion.
• Aussichten für Lambda-Funktion, und wählen Sie die Funktion aus invoke-endpoint-asynch.
• Auswählen Änderungen speichern.

• Öffnen Sie in der AWS Glue-Konsole den Job GlueStreaming-Kinesis-S3.
• Auswählen Job ausführen.

Als nächstes verwenden wir den Kinesis Data Generator (KDG), um Sensoren zu simulieren, die Daten an unseren Kinesis-Datenstrom senden. Wenn Sie das KDG zum ersten Mal verwenden, lesen Sie weiter Überblick für die Ersteinrichtung. Das KDG stellt eine CloudFormation-Vorlage bereit, um den Benutzer zu erstellen und gerade genug Berechtigungen zuzuweisen, um das KDG zum Senden von Ereignissen an Kinesis zu verwenden. Führen Sie die aus CloudFormation-Vorlage innerhalb des AWS-Kontos, das Sie zum Erstellen der Lösung in diesem Beitrag verwenden. Nachdem das KDG eingerichtet ist, melden Sie sich an und greifen Sie auf das KDG zu, um Testereignisse an unseren Kinesis-Datenstrom zu senden.

• Verwenden Sie die Region, in der Sie den Kinesis-Datenstrom erstellt haben (us-east-1).
• Wählen Sie im Dropdown-Menü den Datenstrom aus sensor-data-stream.
• Im Aufzeichnungen pro Sekunde Abschnitt auswählen Konstant und geben Sie 100 ein.
• Auswahl aufheben Datensätze komprimieren.
• Aussichten für Aufnahmevorlage, verwenden Sie die folgende Vorlage:

{
"air_temperature": {{random.number({"min":295,"max":305, "precision":0.01})}},
"process_temperature": {{random.number({"min":305,"max":315, "precision":0.01})}},
"rotational_speed": {{random.number({"min":1150,"max":2900})}},
"torque": {{random.number({"min":3,"max":80, "precision":0.01})}},
"tool_wear": {{random.number({"min":0,"max":250})}},
"type": "{{random.arrayElement(["L","M","H"])}}"
}

• Klicken Sie auf Kostenlos erhalten und dann auf Installieren. Daten senden um mit dem Senden von Daten an den Kinesis-Datenstrom zu beginnen.

Der AWS Glue-Streaming-Auftrag liest und extrahiert basierend auf der bereitgestellten Fenstergröße einen Mikrobatch von Daten (die Sensormesswerte darstellen) aus dem Kinesis-Datenstrom. Der Streaming-Job verarbeitet und führt dann Feature-Engineering für diesen Mikrobatch durch, bevor er partitioniert und in die Präfix-Features im S3-Bucket geschrieben wird.

Wenn neue Funktionen, die vom AWS Glue-Streaming-Auftrag erstellt wurden, in den S3-Bucket geschrieben werden, wird eine Lambda-Funktion (invoke-endpoint-asynch) wird ausgelöst, wodurch ein asynchroner Inferenzendpunkt von SageMaker aufgerufen wird, indem eine Aufrufanforderung gesendet wird, um Inferenzen von unserem bereitgestellten ML-Modell zu erhalten. Der asynchrone Inferenzendpunkt stellt die Anforderung für den asynchronen Aufruf in die Warteschlange. Wenn die Verarbeitung abgeschlossen ist, speichert SageMaker die Inferenzergebnisse im Amazon S3-Speicherort (S3OutputPath), die während der Konfiguration des asynchronen Rückschlussendpunkts angegeben wurde.

Für unseren Anwendungsfall zeigen die Inferenzergebnisse basierend auf den Sensormesswerten an, ob ein Maschinenteil wahrscheinlich ausfällt oder nicht.

SageMaker sendet auch eine Erfolgs- oder Fehlerbenachrichtigung mit Amazon SNS. Wenn Sie zum Beispiel eine E-Mail-Abonnement Für die SNS-Themen „Erfolg“ und „Fehler“ (in der Konfiguration des asynchronen SageMaker-Rückschlussendpunkts angegeben) kann jedes Mal eine E-Mail gesendet werden, wenn eine Rückschlussanforderung verarbeitet wird. Der folgende Screenshot zeigt eine Beispiel-E-Mail aus dem SNS-Erfolgsthema.

Für reale Anwendungen können Sie SNS-Benachrichtigungen mit anderen Diensten integrieren, z Amazon Simple Queue-Dienst (Amazon SQS) und Lambda für die zusätzliche Nachbearbeitung der generierten Inferenzen oder die Integration mit anderen nachgelagerten Anwendungen, basierend auf Ihren Anforderungen. Für unseren Anwendungsfall der vorausschauenden Wartung können Sie beispielsweise eine Lambda-Funktion basierend auf einer SNS-Benachrichtigung aufrufen, um die generierte Inferenz von Amazon S3 zu lesen, weiter zu verarbeiten (z. B. Aggregation oder Filterung) und Workflows wie das Senden von Arbeitsaufträgen zu initiieren Gerätereparatur an Techniker.

Aufräumen

Wenn Sie mit dem Testen des Stacks fertig sind, löschen Sie die Ressourcen (insbesondere den Kinesis-Datenstrom, den Glue-Streaming-Job und die SNS-Themen), um unerwartete Gebühren zu vermeiden.

Führen Sie den folgenden Code aus, um Ihren Stack zu löschen:

sam delete nrt-streaming-inference

Löschen Sie auch die Ressourcen wie SageMaker-Endpunkte, indem Sie dem Bereinigungsabschnitt im ModelTraining-Evaluation-and-Deployment-Notizbuch folgen.

Zusammenfassung

In diesem Beitrag haben wir einen Anwendungsfall für vorausschauende Wartung verwendet, um zu demonstrieren, wie verschiedene Dienste wie Kinesis, AWS Glue und SageMaker verwendet werden können, um eine Inferenzpipeline nahezu in Echtzeit aufzubauen. Wir empfehlen Ihnen, diese Lösung auszuprobieren und uns Ihre Meinung mitzuteilen.

Wenn Sie Fragen haben, teilen Sie sie in den Kommentaren.

Über die Autoren

Rahul Sharma ist ein Lösungsarchitekt bei AWS Data Lab und unterstützt AWS-Kunden beim Entwerfen und Erstellen von KI/ML-Lösungen. Bevor er zu AWS kam, war Rahul mehrere Jahre im Finanz- und Versicherungssektor tätig und half Kunden beim Aufbau von Daten- und Analyseplattformen.

Pat Reilly ist Architekt im AWS Data Lab, wo er Kunden hilft, Daten-Workloads zu entwerfen und zu erstellen, um ihr Geschäft zu unterstützen. Vor AWS beriet sich Pat bei einem AWS-Partner und baute AWS-Daten-Workloads in einer Vielzahl von Branchen auf.