Automatisieren Sie ein gemeinsames Klassifizierungsmodell für Fahrräder und Roller mit Amazon SageMaker Autopilot

Amazon SageMaker-Autopilot ermöglicht es Unternehmen, schnell ein End-to-End-Modell für maschinelles Lernen (ML) und eine Inferenzpipeline mit nur wenigen Codezeilen oder sogar zu erstellen und bereitzustellen ohne Code überhaupt mit Amazon SageMaker-Studio. Autopilot entlastet die schwere Arbeit beim Konfigurieren der Infrastruktur und die Zeit, die zum Erstellen einer gesamten Pipeline benötigt wird, einschließlich Feature-Engineering, Modellauswahl und Hyperparameter-Tuning.

In diesem Beitrag zeigen wir, wie Sie mit Autopilot von Rohdaten zu einer robusten und vollständig bereitgestellten Inferenzpipeline gelangen.

Lösungsüberblick

Wir verwenden Lyfts öffentlicher Datensatz zum Bikesharing damit diese Simulation vorhersagen kann, ob ein Benutzer an der teilnimmt oder nicht Bike Share for All-Programm. Dies ist ein einfaches binäres Klassifizierungsproblem.

Wir möchten zeigen, wie einfach es ist, eine automatisierte Echtzeit-Inferenzpipeline zu erstellen, um Benutzer basierend auf ihrer Teilnahme am Programm „Bike Share for All“ zu klassifizieren. Zu diesem Zweck simulieren wir eine End-to-End-Datenaufnahme- und Inferenzpipeline für ein imaginäres Fahrradverleihunternehmen, das in der San Francisco Bay Area tätig ist.

Die Architektur ist in zwei Teile unterteilt: die Aufnahmepipeline und die Inferenzpipeline.

Wir konzentrieren uns im ersten Abschnitt dieses Beitrags hauptsächlich auf die ML-Pipeline und sehen uns im zweiten Teil die Datenaufnahme-Pipeline an.

Voraussetzungen:

Um diesem Beispiel zu folgen, erfüllen Sie die folgenden Voraussetzungen:

1. Erstellen Sie eine neue SageMaker-Notebookinstanz.
2. Erstellen Sie ein Amazon Kinesis Data Firehose Lieferstrom mit einem AWS Lambda Funktion transformieren. Anweisungen finden Sie unter Amazon Kinesis Firehose-Datentransformation mit AWS Lambda. Dieser Schritt ist optional und wird nur benötigt, um das Datenstreaming zu simulieren.

Datenexploration

Lassen Sie uns den Datensatz herunterladen und visualisieren, der sich in einer öffentlichen Umgebung befindet Amazon Simple Storage-Service (Amazon S3) Bucket und statische Website:

# The dataset is located in a public bucket and static s3 website.
# https://www.lyft.com/bikes/bay-wheels/system-data import pandas as pd
import numpy as np
import os
from time import sleep !wget -q -O '201907-baywheels-tripdata.zip' https://s3.amazonaws.com/baywheels-data/201907-baywheels-tripdata.csv.zip
!unzip -q -o 201907-baywheels-tripdata.zip
csv_file = os.listdir('.')
data = pd.read_csv('201907-baywheels-tripdata.csv', low_memory=False)
data.head()

Der folgende Screenshot zeigt eine Teilmenge der Daten vor der Transformation.

Die letzte Spalte der Daten enthält das Ziel, das wir vorhersagen möchten. Dabei handelt es sich um eine binäre Variable, die entweder einen Ja- oder einen Nein-Wert annimmt und angibt, ob der Benutzer am Bike Share for All-Programm teilnimmt.

Werfen wir einen Blick auf die Verteilung unserer Zielvariablen für ein beliebiges Datenungleichgewicht.

# For plotting
%matplotlib inline
import matplotlib.pyplot as plt
#!pip install seaborn # If you need this library
import seaborn as sns
display(sns.countplot(x='bike_share_for_all_trip', data=data))

Wie in der obigen Grafik dargestellt, sind die Daten unausgewogen, da weniger Personen an dem Programm teilnehmen.

Wir müssen die Daten ausbalancieren, um eine Verzerrung durch Überrepräsentation zu vermeiden. Dieser Schritt ist optional, da Autopilot auch einen internen Ansatz zum automatischen Umgang mit Klassenungleichgewichten bietet, der standardmäßig eine F1-Score-Validierungsmetrik verwendet. Wenn Sie sich entscheiden, die Daten selbst auszugleichen, können Sie außerdem fortgeschrittenere Techniken zum Umgang mit Klassenungleichgewichten verwenden, z SMOTEN or GAN.

Für diesen Beitrag sampeln wir die Mehrheitsklasse (Nein) als Datenausgleichstechnik herunter:

Der folgende Code reichert die Daten an und unterabtastet die überrepräsentierte Klasse:

df = data.copy()
df.drop(columns=['rental_access_method'], inplace=True) df['start_time'] = pd.to_datetime(df['start_time'])
df['start_time'] = pd.to_datetime(df['end_time']) # Adding some day breakdown
df = df.assign(day_of_week=df.start_time.dt.dayofweek, hour_of_day=df.start_time.dt.hour, trip_month=df.start_time.dt.month)
# Breaking the day in 4 parts: ['morning', 'afternoon', 'evening']
conditions = [ (df['hour_of_day'] >= 5) & (df['hour_of_day'] < 12), (df['hour_of_day'] >= 12) & (df['hour_of_day'] < 18), (df['hour_of_day'] >= 18) & (df['hour_of_day'] < 21),
]
choices = ['morning', 'afternoon', 'evening']
df['part_of_day'] = np.select(conditions, choices, default='night')
df.dropna(inplace=True) # Downsampling the majority to rebalance the data
# We are getting about an even distribution
df.sort_values(by='bike_share_for_all_trip', inplace=True)
slice_pointe = int(df['bike_share_for_all_trip'].value_counts()['Yes'] * 2.1)
df = df[-slice_pointe:]
# The data is balanced now. Let's reshuffle the data
df = df.sample(frac=1).reset_index(drop=True)

Wir haben unsere kategorialen Merkmale bewusst nicht codiert gelassen, einschließlich unseres binären Zielwerts. Dies liegt daran, dass Autopilot im Rahmen des automatischen Feature-Engineerings und der Pipeline-Bereitstellung die Codierung und Decodierung der Daten für uns übernimmt, wie wir im nächsten Abschnitt sehen werden.

Der folgende Screenshot zeigt ein Beispiel unserer Daten.

Die Daten in den folgenden Diagrammen sehen ansonsten normal aus, wobei eine bimodale Verteilung die beiden Spitzen für die Morgenstunden und die Stoßzeiten am Nachmittag darstellt, wie Sie es erwarten würden. Wir beobachten auch geringe Aktivitäten an Wochenenden und in der Nacht.

Im nächsten Abschnitt geben wir die Daten an Autopilot weiter, damit dieser ein Experiment für uns durchführen kann.

Erstellen Sie ein binäres Klassifizierungsmodell

Autopilot erfordert, dass wir die Eingabe- und Ausgabeziel-Buckets angeben. Es verwendet den Eingabe-Bucket zum Laden der Daten und den Ausgabe-Bucket zum Speichern der Artefakte, z. B. Feature-Engineering und die generierten Jupyter-Notebooks. Wir behalten 5 % des Datensatzes, um die Leistung des Modells nach Abschluss des Trainings zu bewerten und zu validieren, und laden 95 % des Datensatzes in den S3-Eingabe-Bucket hoch. Siehe folgenden Code:

import sagemaker
import boto3 # Let's define our storage.
# We will use the default sagemaker bucket and will enforce encryption bucket = sagemaker.Session().default_bucket() # SageMaker default bucket. #Encrypting the bucket
s3 = boto3.client('s3')
SSEConfig={ 'Rules': [ { 'ApplyServerSideEncryptionByDefault': { 'SSEAlgorithm': 'AES256', } }, ] }
s3.put_bucket_encryption(Bucket=bucket, ServerSideEncryptionConfiguration=SSEConfig) prefix = 'sagemaker-automl01' # prefix for ther bucket
role = sagemaker.get_execution_role() # IAM role object to use by SageMaker
sagemaker_session = sagemaker.Session() # Sagemaker API
region = sagemaker_session.boto_region_name # AWS Region # Where we will load our data input_path = "s3://{}/{}/automl_bike_train_share-1".format(bucket, prefix) output_path = "s3://{}/{}/automl_bike_output_share-1".format(bucket, prefix) # Spliting data in train/test set.
# We will use 95% of the data for training and the remainder for testing.
slice_point = int(df.shape[0] * 0.95) training_set = df[:slice_point] # 95%
testing_set = df[slice_point:] # 5% # Just making sure we have split it correctly
assert training_set.shape[0] + testing_set.shape[0] == df.shape[0] # Let's save the data locally and upload it to our s3 data location
training_set.to_csv('bike_train.csv')
testing_set.to_csv('bike_test.csv', header=False) # Uploading file the trasining set to the input bucket
sagemaker.s3.S3Uploader.upload(local_path='bike_train.csv', desired_s3_uri=input_path)

Nachdem wir die Daten in das Eingabeziel hochgeladen haben, ist es an der Zeit, Autopilot zu starten:

from sagemaker.automl.automl import AutoML
# You give your job a name and provide the s3 path where you uploaded the data
bike_automl_binary = AutoML(role=role, target_attribute_name='bike_share_for_all_trip', output_path=output_path, max_candidates=30)
# Starting the training bike_automl_binary.fit(inputs=input_path, wait=False, logs=False)

Alles, was wir brauchen, um mit dem Experimentieren zu beginnen, ist, die Methode fit() aufzurufen. Autopilot benötigt als erforderliche Parameter den Ein- und Ausgabe-S3-Standort und die Zielattributspalte. Nach der Funktionsverarbeitung ruft Autopilot auf Automatische Modellabstimmung von SageMaker um die beste Version eines Modells zu finden, indem Sie viele Trainingsjobs auf Ihrem Dataset ausführen. Wir haben den optionalen Parameter max_candidates hinzugefügt, um die Anzahl der Kandidaten auf 30 zu begrenzen, was der Anzahl der Trainingsjobs entspricht, die Autopilot mit verschiedenen Kombinationen von Algorithmen und Hyperparametern startet, um das beste Modell zu finden. Wenn Sie diesen Parameter nicht angeben, wird er standardmäßig auf 250 gesetzt.

Wir können den Fortschritt des Autopiloten mit dem folgenden Code beobachten:

# Let's monitor the progress this will take a while. Go grup some coffe.
from time import sleep def check_job_status(): return bike_automl_binary.describe_auto_ml_job()['AutoMLJobStatus'] def discribe(): return bike_automl_binary.describe_auto_ml_job() while True: print (check_job_status(), discribe()['AutoMLJobSecondaryStatus'], end='** ') if check_job_status() in ["Completed", "Failed"]: if "Failed" in check_job_status(): print(discribe()['FailureReason']) break sleep(20)

Die Ausbildung nimmt einige Zeit in Anspruch. Sehen wir uns während der Ausführung den Autopilot-Workflow an.

Verwenden Sie den folgenden Code, um den besten Kandidaten zu finden:

# Let's take a look at the best candidate selected by AutoPilot
from IPython.display import JSON
def jsonView(obj, rootName=None): return JSON(obj, root=rootName, expanded=True) bestCandidate = bike_automl_binary.describe_auto_ml_job()['BestCandidate']
display(jsonView(bestCandidate['FinalAutoMLJobObjectiveMetric'], 'FinalAutoMLJobObjectiveMetric'))

Der folgende Screenshot zeigt unsere Ausgabe.

Unser Modell hat eine Validierungsgenauigkeit von 96 % erreicht, also werden wir es einsetzen. Wir könnten eine Bedingung hinzufügen, sodass wir das Modell nur verwenden, wenn die Genauigkeit über einem bestimmten Niveau liegt.

Inferenzpipeline

Bevor wir unser Modell einsetzen, sehen wir uns unseren besten Kandidaten an und was in unserer Pipeline vor sich geht. Siehe folgenden Code:

display(jsonView(bestCandidate['InferenceContainers'], 'InferenceContainers'))

Das folgende Diagramm zeigt unsere Ausgabe.

Autopilot hat das Modell erstellt und in drei verschiedene Container gepackt, die jeweils nacheinander eine bestimmte Aufgabe ausführen: Transformieren, Vorhersagen und Rücktransformation. Diese mehrstufige Inferenz ist mit a möglich SageMaker-Inferenzpipeline.

Eine mehrstufige Inferenz kann auch mehrere Inferenzmodelle verketten. Beispielsweise kann ein Container ausgeführt werden Hauptkomponentenanalyse bevor die Daten an den XGBoost-Container übergeben werden.

Stellen Sie die Inferenzpipeline auf einem Endpunkt bereit

Der Bereitstellungsprozess umfasst nur wenige Codezeilen:

# We chose to difine an endpoint name.
from datetime import datetime as dt
today = str(dt.today())[:10]
endpoint_name='binary-bike-share-' + today
endpoint = bike_automl_binary.deploy(initial_instance_count=1, instance_type='ml.m5.xlarge', endpoint_name=endpoint_name, candidate=bestCandidate, wait=True)

Lassen Sie uns unseren Endpunkt für die Vorhersage mit einem Prädiktor konfigurieren:

from sagemaker.serializers import CSVSerializer
from sagemaker.deserializers import CSVDeserializer
csv_serializer = CSVSerializer()
csv_deserializer = CSVDeserializer()
# Initialize the predictor
predictor = sagemaker.predictor.Predictor(endpoint_name=endpoint_name, sagemaker_session=sagemaker.Session(), serializer=csv_serializer, deserializer=csv_deserializer )

Nachdem wir nun unseren Endpunkt und Prädiktor bereit haben, ist es an der Zeit, die Testdaten zu verwenden, die wir beiseite gelegt haben, und die Genauigkeit unseres Modells zu testen. Wir beginnen mit der Definition einer Hilfsfunktion, die die Daten zeilenweise an unseren Inferenzendpunkt sendet und im Gegenzug eine Vorhersage erhält. Denn wir haben eine XGBoost -Modell löschen wir die Zielvariable, bevor wir die CSV-Zeile an den Endpunkt senden. Außerdem haben wir den Header aus der Test-CSV entfernt, bevor wir die Datei durchlaufen, was auch eine weitere Voraussetzung für XGBoost auf SageMaker ist. Siehe folgenden Code:

# The fuction takes 3 arguments: the file containing the test set,
# The predictor and finaly the number of lines to send for prediction.
# The function returns two Series: inferred and Actual.
def get_inference(file, predictor, n=1): infered = [] actual = [] with open(file, 'r') as csv: for i in range(n): line = csv.readline().split(',') #print(line) try: # Here we remove the target variable from the csv line before predicting observed = line.pop(14).strip('n') actual.append(observed) except: pass obj = ','.join(line) predicted = predictor.predict(obj)[0][0] infered.append(predicted) pd.Series(infered) data = {'Infered': pd.Series(infered), 'Observed': pd.Series(actual)} return pd.DataFrame(data=data) n = testing_set.shape[0] # The size of the testing data
inference_df = get_inference('bike_test.csv', predictor, n) inference_df['Binary_Result'] = (inference_df['Observed'] == inference_df['Infered'])
display(inference_df.head())

Der folgende Screenshot zeigt unsere Ausgabe.

Lassen Sie uns nun die Genauigkeit unseres Modells berechnen.

Siehe folgenden Code:

count_binary = inference_df['Binary_Result'].value_counts()
accuracy = count_binary[True]/n
print('Accuracy:', accuracy)

Wir erhalten eine Genauigkeit von 92 %. Dies ist etwas niedriger als die 96 %, die während des Validierungsschritts erhalten wurden, aber immer noch hoch genug. Wir erwarten nicht, dass die Genauigkeit genau gleich ist, da der Test mit einem neuen Datensatz durchgeführt wird.

Datenaufnahme

Wir haben die Daten direkt heruntergeladen und für das Training konfiguriert. Im wirklichen Leben müssen Sie die Daten möglicherweise direkt vom Edge-Gerät in den Data Lake senden und von SageMaker direkt aus dem Data Lake in das Notebook laden lassen.

Kinesis Data Firehose ist eine gute Option und die einfachste Möglichkeit, Streaming-Daten zuverlässig in Data Lakes, Datenspeicher und Analysetools zu laden. Es kann Streaming-Daten erfassen, transformieren und in Amazon S3 und andere AWS-Datenspeicher laden.

Für unseren Anwendungsfall erstellen wir einen Kinesis Data Firehose-Bereitstellungsstream mit einer Lambda-Transformationsfunktion, um beim Durchlaufen des Streams eine einfache Datenbereinigung durchzuführen. Siehe folgenden Code:

# Data processing libraries
import pandas as pd # Data processing
import numpy as np
import base64
from io import StringIO def lambda_handler(event, context): output = [] print('Received', len(event['records']), 'Records') for record in event['records']: payload = base64.b64decode(record['data']).decode('utf-8') df = pd.read_csv(StringIO(payload), index_col=0) df.drop(columns=['rental_access_method'], inplace=True) df['start_time'] = pd.to_datetime(df['start_time']) df['start_time'] = pd.to_datetime(df['end_time']) # Adding some day breakdown df = df.assign(day_of_week=df.start_time.dt.dayofweek, hour_of_day=df.start_time.dt.hour, trip_month=df.start_time.dt.month) # Breaking the day in 4 parts: ['morning', 'afternoon', 'evening'] conditions = [ (df['hour_of_day'] >= 5) & (df['hour_of_day'] < 12), (df['hour_of_day'] >= 12) & (df['hour_of_day'] < 18), (df['hour_of_day'] >= 18) & (df['hour_of_day'] < 21), ] choices = ['morning', 'afternoon', 'evening'] df['part_of_day'] = np.select(conditions, choices, default='night') df.dropna(inplace=True) # Downsampling the majority to rebalance the data # We are getting about an even distribution df.sort_values(by='bike_share_for_all_trip', inplace=True) slice_pointe = int(df['bike_share_for_all_trip'].value_counts()['Yes'] * 2.1) df = df[-slice_pointe:] # The data is balanced now. Let's reshuffle the data df = df.sample(frac=1).reset_index(drop=True) data = base64.b64encode(bytes(df.to_csv(), 'utf-8')).decode("utf-8") output_record = { 'recordId': record['recordId'], 'result': 'Ok', 'data': data } output.append(output_record) print('Returned', len(output), 'Records') print('Event', event) return {'records': output}

Diese Lambda-Funktion führt eine leichte Transformation der Daten durch, die von den Geräten auf den Data Lake gestreamt werden. Es erwartet eine CSV-formatierte Datendatei.

Für den Aufnahmeschritt laden wir die Daten herunter und simulieren einen Datenstrom zu Kinesis Data Firehose mit einer Lambda-Transformationsfunktion und in unseren S3 Data Lake.

Lassen Sie uns das Streaming einiger Zeilen simulieren:

# Saving the data in one file.
file = '201907-baywheels-tripdata.csv' data.to_csv(file) # Stream the data 'n' lines at a time.
# Only run this for a minute and stop the cell
def streamer(file, n): with open(file, 'r') as csvfile: header = next(csvfile) data = header counter = 0 loop = True while loop == True: for i in range(n): line = csvfile.readline() data+=line # We reached the end of the csv file. if line == '': loop = False counter+=n # Use your kinesis streaming name stream = client.put_record(DeliveryStreamName='firehose12-DeliveryStream-OJYW71BPYHF2', Record={"Data": bytes(data, 'utf-8')}) data = header print( file, 'HTTPStatusCode: '+ str(stream['ResponseMetadata']['HTTPStatusCode']), 'csv_lines_sent: ' + str(counter), end=' -*- ') sleep(random.randrange(1, 3)) return
# Streaming for 500 lines at a time. You can change this number up and down.
streamer(file, 500) # We can now load our data as a DataFrame because it’s streamed into the S3 data lake:
# Getting data from s3 location where it was streamed.
STREAMED_DATA = 's3://firehose12-deliverybucket-11z0ya3patrod/firehose/2020'
csv_uri = sagemaker.s3.S3Downloader.list(STREAMED_DATA)
in_memory_string = [sagemaker.s3.S3Downloader.read_file(file) for file in csv_uri]
in_memory_csv = [pd.read_csv(StringIO(file), index_col=0) for file in in_memory_string]
display(df.tail())

Aufräumen

Es ist wichtig, alle in dieser Übung verwendeten Ressourcen zu löschen, um die Kosten zu minimieren. Der folgende Code löscht den von uns erstellten SageMaker-Inferenzendpunkt sowie die von uns hochgeladenen Trainings- und Testdaten:

#Delete the s3 data
predictor.delete_endpoint() # Delete s3 data
s3 = boto3.resource('s3')
ml_bucket = sagemaker.Session().default_bucket()
delete_data = s3.Bucket(ml_bucket).objects.filter(Prefix=prefix).delete()

Zusammenfassung

ML-Ingenieure, Data Scientists und Softwareentwickler können Autopilot verwenden, um eine Inferenzpipeline mit wenig bis gar keiner ML-Programmiererfahrung zu erstellen und bereitzustellen. Autopilot spart Zeit und Ressourcen durch den Einsatz von Best Practices für Data Science und ML. Große Unternehmen können jetzt Engineering-Ressourcen von der Infrastrukturkonfiguration auf die Verbesserung von Modellen und die Lösung von Geschäftsanwendungsfällen verlagern. Startups und kleinere Organisationen können mit wenig bis gar keinem ML-Know-how in maschinelles Lernen einsteigen.

Informationen zu den ersten Schritten mit SageMaker Autopilot finden Sie unter zur Produktseite oder greifen Sie in SageMaker Studio auf SageMaker Autopilot zu.

Wir empfehlen auch, mehr über andere wichtige Funktionen zu erfahren, die SageMaker zu bieten hat, wie z Amazon SageMaker Feature Store, die integriert mit Amazon SageMaker-Pipelines um automatisierte ML-Workflows zu erstellen, Feature-Suche und -Erkennung hinzuzufügen und wiederzuverwenden. Sie können mehrere Autopilot-Simulationen mit unterschiedlichen Feature- oder Zielvarianten in Ihrem Dataset ausführen. Sie können dies auch als dynamisches Fahrzeugzuweisungsproblem angehen, bei dem Ihr Modell versucht, die Fahrzeugnachfrage basierend auf der Zeit (z. B. Tageszeit oder Wochentag) oder dem Standort oder einer Kombination aus beidem vorherzusagen.

Über die Autoren

Doug Mbaya ist Senior Solution Architect mit Schwerpunkt Daten und Analytik. Doug arbeitet eng mit AWS-Partnern zusammen und hilft ihnen bei der Integration von Daten- und Analyselösungen in die Cloud. Zu Dougs früheren Erfahrungen gehört die Unterstützung von AWS-Kunden im Segment Mitfahrgelegenheiten und Essenslieferungen.

Valerio Perrone ist ein Applied Science Manager, der an Amazon SageMaker Automatic Model Tuning und Autopilot arbeitet.

• Coinsmart. Europas beste Bitcoin- und Krypto-Börse.
• Platoblockkette. Web3-Metaverse-Intelligenz. Wissen verstärkt. DEN FREIEN ZUGANG.
• CryptoHawk. Altcoin-Radar. Kostenlose Testphase.
• Quelle: https://aws.amazon.com/blogs/machine-learning/automate-a-shared-bikes-and-scooters-classification-model-with-amazon-sagemaker-autopilot/