Zbuduj rozwiązanie konserwacji predykcyjnej za pomocą Amazon Kinesis, AWS Glue i Amazon SageMaker

Organizacje coraz częściej tworzą i korzystają z rozwiązań opartych na uczeniu maszynowym (ML) dla różnych przypadków użycia i problemów, w tym predykcyjnej konserwacji części maszyn, rekomendacji produktów na podstawie preferencji klientów, profilowania kredytowego, moderacji treści, wykrywania oszustw i nie tylko. W wielu z tych scenariuszy skuteczność i korzyści płynące z rozwiązań opartych na ML można dodatkowo zwiększyć, gdy mogą one przetwarzać i uzyskiwać wgląd w zdarzenia danych w czasie zbliżonym do rzeczywistego.

Chociaż wartość biznesowa i korzyści płynące z rozwiązań opartych na ML w czasie zbliżonym do rzeczywistego są dobrze ugruntowane, architektura wymagana do wdrożenia tych rozwiązań na dużą skalę z optymalną niezawodnością i wydajnością jest skomplikowana. Ten post opisuje, jak możesz łączyć Amazonka Kinesis, Klej AWS, Amazon Sage Maker zbudować rozwiązanie do inżynierii funkcji i wnioskowania w czasie zbliżonym do rzeczywistego na potrzeby konserwacji predykcyjnej.

Omówienie przypadków użycia

Koncentrujemy się na przypadku użycia konserwacji predykcyjnej, w którym czujniki rozmieszczone w terenie (takie jak sprzęt przemysłowy lub urządzenia sieciowe) wymagają wymiany lub naprawy, zanim ulegną uszkodzeniu i spowodują przestój. Przestój może być kosztowny dla firm i może prowadzić do niskiej jakości obsługi klienta. Konserwacja predykcyjna oparta na modelu ML może również pomóc w rozszerzeniu regularnych cykli konserwacji opartych na harmonogramie, informując, kiedy część maszyny w dobrym stanie nie powinna być wymieniana, co pozwala uniknąć niepotrzebnych kosztów.

W tym poście koncentrujemy się na zastosowaniu uczenia maszynowego do syntetycznego zestawu danych zawierającego awarie maszyn spowodowane takimi cechami, jak temperatura powietrza, temperatura procesu, prędkość obrotowa, moment obrotowy i zużycie narzędzia. Użyty zbiór danych pochodzi z Repozytorium danych UCI.

Awaria maszyny składa się z pięciu niezależnych trybów awarii:

• Awaria zużycia narzędzia (TWF)
• Awaria rozpraszania ciepła (HDF)
• Awaria zasilania (PWF)
• Awaria przeciążenia (OSF)
• Awaria losowa (RNF)

Etykieta awarii maszyny wskazuje, czy maszyna uległa awarii dla określonego punktu danych, jeśli którykolwiek z poprzednich trybów awarii jest prawdziwy. Jeśli co najmniej jeden z trybów awarii jest prawdziwy, proces kończy się niepowodzeniem, a etykieta awarii maszyny jest ustawiona na 1. Celem modelu ML jest prawidłowa identyfikacja awarii maszyny, aby można było zainicjować działanie predykcyjnej konserwacji.

Omówienie rozwiązania

W przypadku konserwacji predykcyjnej zakładamy, że czujniki urządzeń przesyłają różne pomiary i odczyty dotyczące części maszyn. Nasze rozwiązanie za każdym razem pobiera wycinek danych przesyłanych strumieniowo (mikropartie) i wykonuje przetwarzanie oraz inżynierię funkcji w celu tworzenia funkcji. Utworzone funkcje są następnie wykorzystywane do generowania wniosków z wytrenowanego i wdrożonego modelu ML w czasie zbliżonym do rzeczywistego. Wygenerowane wnioski mogą być dalej przetwarzane i wykorzystywane przez dalsze aplikacje w celu podjęcia odpowiednich działań i zainicjowania czynności konserwacyjnych.

Poniższy diagram przedstawia architekturę naszego ogólnego rozwiązania.

Rozwiązanie zasadniczo składa się z następujących sekcji, które zostały szczegółowo wyjaśnione w dalszej części tego postu:

• Strumieniowe źródło i pozyskiwanie danych - Używamy Strumienie danych Amazon Kinesis do zbierania danych strumieniowych z czujników polowych na dużą skalę i udostępniania ich do dalszego przetwarzania.
• Inżynieria funkcji w czasie zbliżonym do rzeczywistego – Używamy zadań przesyłania strumieniowego AWS Glue do odczytywania danych ze strumienia danych Kinesis oraz wykonywania przetwarzania danych i inżynierii funkcji, przed przechowywaniem pochodnych funkcji w Usługa Amazon Simple Storage (Amazonka S3). Amazon S3 zapewnia niezawodną i ekonomiczną opcję przechowywania dużych ilości danych.
• Szkolenie i wdrażanie modeli – Używamy zestawu danych konserwacji predykcyjnej AI4I z repozytorium danych UCI do trenowania modelu ML opartego na algorytmie XGBoost przy użyciu programu SageMaker. Następnie wdrażamy wyszkolony model w punkcie końcowym wnioskowania asynchronicznego SageMaker.
• Wnioskowanie ML w czasie zbliżonym do rzeczywistego – Po udostępnieniu funkcji w Amazon S3 musimy generować wnioski z wdrożonego modelu w czasie zbliżonym do rzeczywistego. Punkty końcowe wnioskowania asynchronicznego programu SageMaker są dobrze dostosowane do tego wymagania, ponieważ obsługują większe rozmiary ładunku (do 1 GB) i mogą generować wnioskowania w ciągu kilku minut (maksymalnie do 15 minut). Używamy powiadomień o zdarzeniach S3, aby uruchomić AWS Lambda funkcja do wywoływania punktu końcowego wnioskowania asynchronicznego SageMaker. Asynchroniczne punkty końcowe wnioskowania SageMaker akceptują lokalizacje S3 jako dane wejściowe, generują wnioski z wdrożonego modelu i zapisują te wnioski z powrotem do Amazon S3 w czasie zbliżonym do rzeczywistego.

Kod źródłowy tego rozwiązania znajduje się na GitHub. Rozwiązanie zostało przetestowane i powinno działać w us-east-1.

Używamy Tworzenie chmury AWS szablon, wdrożony przy użyciu AWS Serverless Model aplikacji (AWS SAM) i notebooki SageMaker w celu wdrożenia rozwiązania.

Wymagania wstępne

Aby rozpocząć, jako warunek wstępny musisz mieć CLI SAM, Python 3, PIP zainstalowany. Musisz też mieć Interfejs wiersza poleceń AWS (AWS CLI) skonfigurowane poprawnie.

Wdróż rozwiązanie

Możesz użyć Chmura AWS aby wykonać te kroki. CloudShell to powłoka oparta na przeglądarce, która jest wstępnie uwierzytelniana przy użyciu poświadczeń konsoli i zawiera wstępnie zainstalowane wspólne narzędzia programistyczne i operacyjne (takie jak AWS SAM, AWS CLI i Python). Dlatego nie jest wymagana lokalna instalacja ani konfiguracja.

• Zaczynamy od utworzenia wiadra S3, w którym przechowujemy skrypt dla naszego zadania przesyłania strumieniowego AWS Glue. Uruchom w terminalu następujące polecenie, aby utworzyć nowy zasobnik:

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

• Zanotuj nazwę utworzonego zasobnika.

• Następnie sklonujemy lokalnie repozytorium kodu, które zawiera szablon CloudFormation do wdrożenia stosu. Uruchom następujące polecenie w swoim terminalu:

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

• Przejdź do katalogu sam-template:

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

• Uruchom następujące polecenie, aby skopiować skrypt zadania AWS Glue (z pliku glue_streaming/app.py) do utworzonego zasobnika S3:

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

• Możesz teraz kontynuować budowanie i wdrażanie rozwiązania za pomocą szablonu CloudFormation za pośrednictwem AWS SAM. Uruchom następujące polecenie:

sam deploy --guided

• Podaj argumenty dla wdrożenia, takie jak nazwa stosu, preferowany region AWS (us-east-1), A GlueScriptsBucket.

Upewnij się, że podajesz to samo wiadro S3, które utworzyłeś wcześniej dla wiaderka S3 skryptu AWS Glue (parametr GlueScriptsBucket na poniższym zrzucie ekranu).

Po podaniu wymaganych argumentów AWS SAM rozpocznie wdrażanie stosu. Poniższy zrzut ekranu przedstawia utworzone zasoby.

Po pomyślnym wdrożeniu stosu powinien zostać wyświetlony następujący komunikat.

• W konsoli AWS CloudFormation otwórz stos (dla tego posta, nrt-streaming-inference), który został podany podczas wdrażania szablonu CloudFormation.
• Na Zasoby zanotuj identyfikator instancji notatnika SageMaker.

• W konsoli SageMaker otwórz tę instancję.

Instancja notatnika SageMaker ma już wstępnie załadowane wymagane notatniki.

Przejdź do folderu notatników i otwórz i postępuj zgodnie z instrukcjami w notatnikach (Data_Pre-Processing.ipynb i ModelTraining-Evaluation-and-Deployment.ipynb), aby eksplorować zestaw danych, wykonywać wstępne przetwarzanie i inżynierię funkcji oraz trenować i wdrażać model w punkcie końcowym wnioskowania asynchronicznego programu SageMaker.

Strumieniowe źródło i pozyskiwanie danych

Kinesis Data Streams to bezserwerowa, skalowalna i trwała usługa przesyłania strumieniowego danych w czasie rzeczywistym, której można używać do zbierania i przetwarzania dużych strumieni rekordów danych w czasie rzeczywistym. Kinesis Data Streams umożliwia przechwytywanie, przetwarzanie i przechowywanie strumieni danych z różnych źródeł, takich jak dane dziennika infrastruktury IT, dzienniki aplikacji, media społecznościowe, kanały danych rynkowych, dane strumienia kliknięć w sieci, urządzenia i czujniki IoT i inne. Strumień danych Kinesis można aprowizować w trybie na żądanie lub w trybie aprowizowanym w zależności od wymagań dotyczących przepustowości i skalowania. Aby uzyskać więcej informacji, zobacz Wybór trybu pojemności strumienia danych.

W naszym przypadku użycia zakładamy, że różne czujniki wysyłają do strumienia danych pomiary, takie jak temperatura, prędkość obrotowa, moment obrotowy i zużycie narzędzia. Strumienie danych Kinesis działają jak lejek do zbierania i przyjmowania strumieni danych.

Używamy Generator danych Amazon Kinesis (KDG) w dalszej części tego postu, aby wygenerować i wysłać dane do strumienia danych Kinesis, symulując dane generowane przez czujniki. Dane ze strumienia danych strumienia danych czujnika są pozyskiwane i przetwarzane przy użyciu zadania przesyłania strumieniowego AWS Glue, które omówimy dalej.

Inżynieria funkcji w czasie zbliżonym do rzeczywistego

Zadania przesyłania strumieniowego kleju AWS zapewniają wygodny sposób przetwarzania danych przesyłanych strumieniowo na dużą skalę, bez konieczności zarządzania środowiskiem obliczeniowym. AWS Glue umożliwia wykonywanie operacji wyodrębniania, przekształcania i ładowania (ETL) na przesyłanych strumieniowo danych przy użyciu stale działających zadań. AWS Glue streaming ETL jest zbudowany na silniku Apache Spark Structured Streaming i może pozyskiwać strumienie z Kinesis, Apache Kafka i Przesyłanie strumieniowe zarządzane przez Amazon dla Apache Kafka (Amazon MSK).

Zadanie przesyłania strumieniowego ETL może korzystać zarówno z wbudowanych przekształceń AWS Glue, jak i przekształceń, które są natywne dla Apache Spark Structured Streaming. Możesz także użyć Spark ML i MLLib biblioteki w zadaniach AWS Glue dla łatwiejszego przetwarzania funkcji przy użyciu łatwo dostępnych bibliotek pomocniczych.

Jeśli schemat źródła danych przesyłania strumieniowego jest wstępnie określony, można go określić w tabeli wykazu danych AWS. Jeśli nie można wcześniej określić definicji schematu, można włączyć wykrywanie schematu w zadaniu przesyłania strumieniowego ETL. Zadanie następnie automatycznie określa schemat na podstawie przychodzących danych. Dodatkowo możesz użyć Rejestr schematów kleju AWS aby umożliwić centralne wykrywanie, kontrolę i ewolucję schematów strumieni danych. Możesz dalej zintegrować Rejestr schematów z Katalogiem danych, aby opcjonalnie używać schematów przechowywanych w Rejestrze schematów podczas tworzenia lub aktualizowania tabel lub partycji AWS Glue w Katalogu danych.

W tym poście tworzymy tabelę Katalogu danych kleju AWS (sensor-stream) z naszym strumieniem danych Kinesis jako źródłem i zdefiniuj schemat dla naszych danych czujnika.

Tworzymy dynamiczną ramkę danych AWS Glue z tabeli Data Catalog, aby odczytać dane strumieniowe z Kinesis. Określamy również następujące opcje:

• Rozmiar okna wynoszący 60 sekund, dzięki czemu zadanie AWS Glue odczytuje i przetwarza dane w 60-sekundowych oknach
• Pozycja wyjściowa TRIM_HORIZON, aby umożliwić odczyt najstarszych rekordów w strumieniu danych Kinesis

Używamy również bibliotek Spark MLlib Indeksator ciągów transformator funkcji do kodowania typu kolumny ciągu w indeksach etykiet. Ta transformacja jest implementowana przy użyciu Spark ML Pipelines. Rurociągi Spark ML zapewnić jednolity zestaw interfejsów API wysokiego poziomu dla algorytmów ML, aby ułatwić łączenie wielu algorytmów w jeden potok lub przepływ pracy.

Używamy interfejsu API foreachBatch do wywołania funkcji o nazwie processBatch, która z kolei przetwarza dane, do których odwołuje się ta ramka danych. Zobacz następujący kod:

# 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})

Funkcja processBatch wykonuje określone przekształcenia i partycjonuje dane w Amazon S3 na podstawie roku, miesiąca, dnia i identyfikatora partii.

Ponownie dzielimy partycje AWS Glue na pojedynczą partycję, aby uniknąć zbyt wielu małych plików w Amazon S3. Posiadanie kilku małych plików może zmniejszać wydajność odczytu, ponieważ zwiększa obciążenie związane z wyszukiwaniem, otwieraniem i odczytywaniem każdego pliku. Na koniec piszemy funkcje do generowania wnioskowań w prefiksie (funkcjach) w zasobniku S3. Zobacz następujący kod:

# 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()))

Modeluj szkolenie i wdrażanie

SageMaker to w pełni zarządzana i zintegrowana usługa ML, która umożliwia analitykom danych i inżynierom ML szybkie i łatwe tworzenie, szkolenie i wdrażanie modeli ML.

W notatniku Data_Pre-Processing.ipynb, najpierw importujemy zestaw danych AI4I Predictive Maintenance z repozytorium danych UCI i przeprowadzamy eksploracyjną analizę danych (EDA). Wykonujemy również inżynierię funkcji, aby nasze funkcje były bardziej przydatne do uczenia modelu.

Na przykład w zestawie danych mamy funkcję o nazwie typ, która reprezentuje typ jakości produktu jako L (niska), M (średnia) lub H (wysoka). Ponieważ jest to cecha kategoryczna, musimy ją zakodować przed trenowaniem naszego modelu. Aby to osiągnąć, używamy LabelEncoder firmy Scikit-Learn:

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

Po przetworzeniu funkcji i wygenerowaniu wyselekcjonowanego pociągu i testowych zestawów danych jesteśmy gotowi do trenowania modelu ML, aby przewidzieć, czy maszyna uległa awarii, czy nie, na podstawie odczytów systemu. Trenujemy model XGBoost, korzystając z wbudowanego algorytmu SageMaker. XGBoost może zapewnić dobre wyniki dla wielu typów problemów ML, w tym klasyfikacji, nawet jeśli próbki szkoleniowe są ograniczone.

Oferty pracy szkoleniowej SageMaker zapewniają potężny i elastyczny sposób uczenia modeli ML w programie SageMaker. SageMaker zarządza podstawową infrastrukturą obliczeniową i zapewnia wiele opcji do wyboru, dla różnych wymagań szkoleniowych modelu, w oparciu o przypadek użycia.

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})

Gdy szkolenie modelu jest zakończone, a ocena modelu jest satysfakcjonująca na podstawie wymagań biznesowych, możemy rozpocząć wdrażanie modelu. Najpierw tworzymy konfigurację punktu końcowego z opcją obiektu AsyncInferenceConfig i przy użyciu wytrenowanego wcześniej modelu:

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},
},)

Następnie tworzymy punkt końcowy asynchronicznego wnioskowania SageMaker, używając utworzonej przez nas konfiguracji punktu końcowego. Po aprowizacji możemy rozpocząć wywoływanie punktu końcowego, aby asynchronicznie generować wnioskowania.

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

Wnioskowanie w czasie zbliżonym do rzeczywistego

SageMaker wnioskowanie asynchroniczne punkty końcowe zapewniają możliwość kolejkowania przychodzących żądań wnioskowania i przetwarzania ich asynchronicznie w czasie zbliżonym do rzeczywistego. Jest to idealne rozwiązanie w przypadku aplikacji, które mają żądania wnioskowania z większymi rozmiarami ładunku (do 1 GB), mogą wymagać dłuższego czasu przetwarzania (do 15 minut) i mają wymagania dotyczące opóźnień w czasie zbliżonym do rzeczywistego. Wnioskowanie asynchroniczne pozwala również zaoszczędzić na kosztach przez automatyczne skalowanie liczby wystąpień do zera, gdy nie ma żądań do przetworzenia, więc płacisz tylko wtedy, gdy punkt końcowy przetwarza żądania.

Możesz utworzyć punkt końcowy wnioskowania asynchronicznego SageMaker podobnie do tego, jak tworzysz punkt końcowy wnioskowania w czasie rzeczywistym i dodatkowo określić AsyncInferenceConfig obiektu, podczas tworzenia konfiguracji punktu końcowego za pomocą pola EndpointConfig w interfejsie API CreateEndpointConfig. Poniższy diagram przedstawia przepływ pracy wnioskowania i sposób, w jaki asynchroniczny punkt końcowy wnioskowania generuje wnioskowanie.

Aby wywołać punkt końcowy wnioskowania asynchronicznego, ładunek żądania powinien być przechowywany w Amazon S3, a odwołanie do tego ładunku musi być dostarczone w ramach żądania InvokeEndpointAsync. Po wywołaniu SageMaker kolejkuje żądanie do przetworzenia i zwraca w odpowiedzi identyfikator oraz lokalizację wyjściową. Po przetworzeniu SageMaker umieszcza wynik w lokalizacji Amazon S3. Możesz opcjonalnie wybrać otrzymywanie powiadomień o powodzeniu lub błędach za pomocą Usługa prostego powiadomienia Amazon (Amazonskie SNS).

Przetestuj kompleksowe rozwiązanie

Aby przetestować rozwiązanie, wykonaj następujące czynności:

• W konsoli AWS CloudFormation otwórz utworzony wcześniej stos (nrt-streaming-inference).
• Na Wyjścia skopiuj nazwę kubełka S3 (EventsBucket).

Jest to wiadro S3, do którego nasze zadanie przesyłania strumieniowego AWS Glue zapisuje funkcje po odczytaniu i przetworzeniu ze strumienia danych Kinesis.

Następnie konfigurujemy powiadomienia o zdarzeniach dla tego zasobnika S3.

• W konsoli Amazon S3 przejdź do zasobnika EventsBucket.
• Na Właściwości zakładka, w Powiadomienia o wydarzeniach Sekcja, wybierz Utwórz powiadomienie o wydarzeniu.

• W razie zamówieenia projektu Nazwa wydarzenia, wchodzić invoke-endpoint-lambda.
• W razie zamówieenia projektu Prefiks, wchodzić features/.
• W razie zamówieenia projektu Przyrostek, wchodzić .csv.
• W razie zamówieenia projektu Rodzaje wydarzeń, Wybierz Wszystkie obiekty tworzą zdarzenia.

• W razie zamówieenia projektu Cel, Wybierz Funkcja Lambdy.
• W razie zamówieenia projektu Funkcja Lambdyi wybierz funkcję invoke-endpoint-asynch.
• Dodaj Zapisz zmiany.

• W konsoli AWS Glue otwórz zadanie GlueStreaming-Kinesis-S3.
• Dodaj Uruchom zadanie.

Następnie używamy Kinesis Data Generator (KDG) do symulacji czujników wysyłających dane do naszego strumienia danych Kinesis. Jeśli po raz pierwszy używasz KDG, zapoznaj się z Przegląd dla początkowej konfiguracji. KDG udostępnia szablon CloudFormation do tworzenia użytkownika i przypisywania wystarczających uprawnień, aby używać KDG do wysyłania zdarzeń do Kinesis. Uruchom Szablon CloudFormation w ramach konta AWS, którego używasz do tworzenia rozwiązania w tym poście. Po skonfigurowaniu KDG zaloguj się i uzyskaj dostęp do KDG, aby wysłać zdarzenia testowe do naszego strumienia danych Kinesis.

• Użyj regionu, w którym utworzono strumień danych Kinesis (us-east-1).
• Z menu rozwijanego wybierz strumień danych sensor-data-stream.
• W Rekordy na sekundę sekcja, wybierz stały i wejdź w 100.
• Odznacz Kompresuj rekordy.
• W razie zamówieenia projektu Szablon zapisu, użyj następującego szablonu:

{
"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"])}}"
}

• Kliknij Przesłać dane aby rozpocząć wysyłanie danych do strumienia danych Kinesis.

Zadanie przesyłania strumieniowego AWS Glue odczytuje i wyodrębnia mikropartię danych (reprezentujących odczyty czujnika) ze strumienia danych Kinesis na podstawie dostarczonego rozmiaru okna. Zadanie przesyłania strumieniowego następnie przetwarza i wykonuje inżynierię funkcji na tej mikropartii przed partycjonowaniem i zapisaniem jej do funkcji prefiksu w zasobniku S3.

Ponieważ nowe funkcje tworzone przez zadanie przesyłania strumieniowego AWS Glue są zapisywane w wiadrze S3, funkcja Lambda (invoke-endpoint-asynch) jest wyzwalane, co wywołuje punkt końcowy asynchronicznego wnioskowania SageMaker, wysyłając żądanie wywołania w celu uzyskania wnioskowań z naszego wdrożonego modelu ML. Punkt końcowy wnioskowania asynchronicznego kolejkuje żądanie wywołania asynchronicznego. Po zakończeniu przetwarzania SageMaker przechowuje wyniki wnioskowania w lokalizacji Amazon S3 (S3OutputPath), który został określony podczas konfiguracji punktu końcowego wnioskowania asynchronicznego.

W naszym przypadku użycia wyniki wnioskowania wskazują, czy część maszyny prawdopodobnie ulegnie awarii, czy nie, na podstawie odczytów czujnika.

SageMaker wysyła również powiadomienie o powodzeniu lub błędzie za pomocą Amazon SNS. Na przykład, jeśli skonfigurujesz Subskrypcja e-mailowa w przypadku tematów SNS powodzenia i błędu (określonych w konfiguracji punktu końcowego wnioskowania asynchronicznego programu SageMaker) można wysłać wiadomość e-mail za każdym razem, gdy przetwarzane jest żądanie wnioskowania. Poniższy zrzut ekranu przedstawia przykładową wiadomość e-mail z tematu sukcesu SNS.

W przypadku rzeczywistych aplikacji możesz zintegrować powiadomienia SNS z innymi usługami, takimi jak Usługa Amazon Simple Queue (Amazon SQS) i Lambda do dodatkowego przetwarzania końcowego wygenerowanych wniosków lub integracji z innymi dalszymi aplikacjami, w oparciu o Twoje wymagania. Na przykład w naszym przypadku użycia konserwacji predykcyjnej można wywołać funkcję Lambda opartą na powiadomieniu SNS w celu odczytania wygenerowanego wnioskowania z Amazon S3, dalszego jego przetwarzania (np. agregacji lub filtrowania) i inicjowania przepływów pracy, takich jak wysyłanie zleceń pracy dla naprawa sprzętu dla techników.

Sprzątać

Po zakończeniu testowania stosu usuń zasoby (zwłaszcza strumień danych Kinesis, zadanie przesyłania strumieniowego kleju i tematy SNS), aby uniknąć nieoczekiwanych opłat.

Uruchom następujący kod, aby usunąć swój stos:

sam delete nrt-streaming-inference

Usuń również zasoby, takie jak punkty końcowe SageMaker, postępując zgodnie z sekcją czyszczenia w notesie ModelTraining-Evaluation-and-Deployment.

Wnioski

W tym poście wykorzystaliśmy przypadek użycia konserwacji predykcyjnej, aby zademonstrować, jak korzystać z różnych usług, takich jak Kinesis, AWS Glue i SageMaker, w celu zbudowania potoku wnioskowania w czasie zbliżonym do rzeczywistego. Zachęcamy do wypróbowania tego rozwiązania i daj nam znać, co myślisz.

Jeśli masz jakieś pytania, podziel się nimi w komentarzach.

O autorach

Rahul Sharma jest Architektem Rozwiązań w AWS Data Lab, pomagającym klientom AWS projektować i budować rozwiązania AI/ML. Przed dołączeniem do AWS Rahul spędził kilka lat w sektorze finansowym i ubezpieczeniowym, pomagając klientom budować platformy danych i analityczne.

Pata Reilly'ego jest architektem w AWS Data Lab, gdzie pomaga klientom projektować i budować obciążenia danych wspierające ich biznes. Przed AWS Pat konsultował się z partnerem AWS, budując obciążenia danych AWS w różnych branżach.