Zautomatyzuj współdzielony model klasyfikacji rowerów i skuterów za pomocą Amazon SageMaker Autopilot

Autopilot Amazon SageMaker umożliwia organizacjom szybkie budowanie i wdrażanie kompleksowego modelu uczenia maszynowego (ML) i potoku wnioskowania za pomocą zaledwie kilku linijek kodu lub nawet bez żadnego kodu w ogóle z Studio Amazon SageMaker. Autopilot odciąża ciężką konfigurację infrastruktury i czas potrzebny na zbudowanie całego potoku, w tym inżynierię funkcji, wybór modelu i dostrajanie hiperparametrów.

W tym poście pokazujemy, jak przejść od surowych danych do solidnego i w pełni wdrożonego potoku wnioskowania za pomocą rozwiązania Autopilot.

Omówienie rozwiązania

Używamy pliki Publiczny zbiór danych Lyft na temat rowerów publicznych aby ta symulacja mogła przewidzieć, czy użytkownik uczestniczy w Program „Udostępnij rower dla wszystkich”.. Jest to prosty problem klasyfikacji binarnej.

Chcemy pokazać, jak łatwo można zbudować zautomatyzowany potok wnioskowania działający w czasie rzeczywistym i klasyfikujący użytkowników na podstawie ich udziału w programie Bike Share for All. W tym celu symulujemy kompleksowe pozyskiwanie danych i potok wnioskowania dla wyimaginowanej firmy zajmującej się wypożyczaniem rowerów działającej w rejonie Zatoki San Francisco.

Architektura jest podzielona na dwie części: potok pozyskiwania i potok wnioskowania.

W pierwszej części tego posta skupimy się przede wszystkim na potoku uczenia maszynowego, a w drugiej części omówimy potok pozyskiwania danych.

Wymagania wstępne

Aby postępować zgodnie z tym przykładem, należy spełnić następujące wymagania wstępne:

1. Utwórz nową instancję notatnika SageMaker.
2. Tworzenie Wąż strażacki Amazon Kinesis Data strumień dostawy z AWS Lambda funkcja transformacji. Aby uzyskać instrukcje, zobacz Transformacja danych Amazon Kinesis Firehose za pomocą AWS Lambda. Ten krok jest opcjonalny i potrzebny tylko do symulacji przesyłania strumieniowego danych.

Eksploracja danych

Pobierzmy i zwizualizujmy zbiór danych, który znajduje się w przestrzeni publicznej Usługa Amazon Simple Storage Wiadro (Amazon S3) i statyczna strona internetowa:

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

Poniższy zrzut ekranu przedstawia podzbiór danych przed transformacją.

Ostatnia kolumna danych zawiera cel, który chcemy przewidzieć, czyli zmienną binarną przyjmującą wartość Tak lub Nie, wskazującą, czy użytkownik uczestniczy w programie Bike Share for All.

Przyjrzyjmy się rozkładowi naszej zmiennej docelowej dla dowolnej nierównowagi danych.

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

Jak widać na powyższym wykresie, dane są niezrównoważone – w programie uczestniczy mniej osób.

Musimy zrównoważyć dane, aby zapobiec stronniczości związanej z nadreprezentacją. Ten krok jest opcjonalny, ponieważ rozwiązanie Autopilot oferuje również wewnętrzne podejście do automatycznego radzenia sobie z brakiem równowagi klas, co domyślnie jest metryką sprawdzania poprawności wyniku F1. Dodatkowo, jeśli zdecydujesz się samodzielnie zrównoważyć dane, możesz użyć bardziej zaawansowanych technik radzenia sobie z brakiem równowagi klas, takich jak GŁADKIE or GAN.

W tym poście zmniejszamy klasę większości (Nie) jako technikę równoważenia danych:

Poniższy kod wzbogaca dane i zaniża próbkę nadreprezentowanej klasy:

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)

Celowo nie zakodowaliśmy naszych cech kategorycznych, w tym naszej binarnej wartości docelowej. Dzieje się tak, ponieważ Autopilot zajmuje się za nas kodowaniem i dekodowaniem danych w ramach automatycznego projektowania funkcji i wdrażania potoków, jak zobaczymy w następnej sekcji.

Poniższy zrzut ekranu przedstawia próbkę naszych danych.

Dane na poniższych wykresach wyglądają normalnie, a rozkład bimodalny przedstawia dwa szczyty w godzinach porannych i popołudniowych godzinach szczytu, jak można się spodziewać. Niską aktywność obserwujemy również w weekendy i w nocy.

W następnej sekcji przekazujemy dane Autopilotowi, aby mógł przeprowadzić za nas eksperyment.

Zbuduj binarny model klasyfikacji

Autopilot wymaga określenia docelowych segmentów wejściowych i wyjściowych. Używa segmentu wejściowego do ładowania danych i segmentu wyjściowego do zapisywania artefaktów, takich jak inżynieria funkcji i wygenerowane notesy Jupyter. Zachowujemy 5% zbioru danych w celu oceny i sprawdzenia wydajności modelu po zakończeniu szkolenia i przesyłamy 95% zbioru danych do segmentu wejściowego S3. Zobacz następujący kod:

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)

Po przesłaniu danych do miejsca docelowego wejściowego czas uruchomić Autopilota:

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)

Aby rozpocząć eksperymenty, wystarczy wywołać metodę fit(). Autopilot potrzebuje wejściowej i wyjściowej lokalizacji S3 oraz docelowej kolumny atrybutów jako wymaganych parametrów. Po przetworzeniu funkcji następuje wywołanie Autopilota Automatyczne dostrajanie modelu SageMaker aby znaleźć najlepszą wersję modelu, uruchamiając wiele zadań szkoleniowych na zestawie danych. Dodaliśmy opcjonalny parametr max_candidates, aby ograniczyć liczbę kandydatów do 30, czyli liczby zadań szkoleniowych, które Autopilot uruchamia z różnymi kombinacjami algorytmów i hiperparametrów w celu znalezienia najlepszego modelu. Jeśli nie określisz tego parametru, jego wartość domyślna to 250.

Postęp Autopilota możemy obserwować za pomocą następującego kodu:

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

Ukończenie szkolenia zajmuje trochę czasu. Podczas jego działania przyjrzyjmy się przepływowi pracy autopilota.

Aby znaleźć najlepszego kandydata użyj poniższego kodu:

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

Poniższy zrzut ekranu przedstawia nasze dane wyjściowe.

Nasz model osiągnął dokładność walidacji na poziomie 96%, dlatego zamierzamy go wdrożyć. Moglibyśmy dodać warunek, że będziemy używać modelu tylko wtedy, gdy dokładność przekracza określony poziom.

Potok wnioskowania

Zanim wdrożymy nasz model, przeanalizujmy naszego najlepszego kandydata i to, co dzieje się w przygotowaniu. Zobacz następujący kod:

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

Poniższy diagram przedstawia nasze dane wyjściowe.

Autopilot zbudował model i umieścił go w trzech różnych kontenerach, z których każdy wykonuje po kolei określone zadanie: transformację, przewidywanie i transformację odwrotną. To wieloetapowe wnioskowanie jest możliwe za pomocą a Potok wnioskowania SageMaker.

Wnioskowanie wieloetapowe może również łączyć wiele modeli wnioskowania. Na przykład jeden kontener może wykonać analiza głównego składnika przed przekazaniem danych do kontenera XGBoost.

Wdróż potok wnioskowania w punkcie końcowym

Proces wdrażania obejmuje tylko kilka linii kodu:

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

Skonfigurujmy nasz punkt końcowy do przewidywania za pomocą predyktora:

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 )

Teraz, gdy mamy już gotowy punkt końcowy i predyktor, czas skorzystać z danych testowych, które odłożyliśmy i przetestować dokładność naszego modelu. Zaczynamy od zdefiniowania funkcji użyteczności, która wysyła dane pojedynczo do punktu końcowego wnioskowania i w zamian otrzymuje prognozę. Ponieważ mamy XGBoost modelu, usuwamy zmienną docelową przed wysłaniem linii CSV do punktu końcowego. Dodatkowo usunęliśmy nagłówek z testowego pliku CSV przed przeglądaniem pliku w pętli, co jest również kolejnym wymaganiem XGBoost w SageMaker. Zobacz następujący kod:

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

Poniższy zrzut ekranu przedstawia nasze dane wyjściowe.

Obliczmy teraz dokładność naszego modelu.

Zobacz następujący kod:

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

Otrzymujemy dokładność na poziomie 92%. Jest to nieco mniej niż 96% uzyskane na etapie walidacji, ale nadal jest wystarczająco wysokie. Nie oczekujemy, że dokładność będzie dokładnie taka sama, ponieważ test jest wykonywany na nowym zbiorze danych.

Pozyskiwanie danych

Pobraliśmy dane bezpośrednio i skonfigurowaliśmy je do szkolenia. W prawdziwym życiu może zaistnieć konieczność wysłania danych bezpośrednio z urządzenia brzegowego do jeziora danych i pobrania ich przez SageMaker bezpośrednio z jeziora danych do notebooka.

Kinesis Data Firehose to dobra opcja i najprostszy sposób niezawodnego ładowania danych przesyłanych strumieniowo do jezior danych, magazynów danych i narzędzi analitycznych. Może przechwytywać, przekształcać i ładować dane przesyłane strumieniowo do Amazon S3 i innych magazynów danych AWS.

W naszym przypadku tworzymy strumień dostarczający Kinesis Data Firehose z funkcją transformacji Lambda, aby wykonać lekkie czyszczenie danych podczas ich przechodzenia przez strumień. Zobacz następujący kod:

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

Ta funkcja Lambda dokonuje lekkiej transformacji danych przesyłanych strumieniowo z urządzeń do jeziora danych. Oczekuje pliku danych w formacie CSV.

Na etapie przetwarzania pobieramy dane i symulujemy strumień danych do Kinesis Data Firehose z funkcją transformacji Lambda oraz do naszego jeziora danych S3.

Zasymulujmy przesyłanie strumieniowe kilku linii:

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

Sprzątać

Aby zminimalizować koszty, ważne jest usunięcie wszystkich zasobów używanych w tym ćwiczeniu. Poniższy kod usuwa utworzony przez nas punkt końcowy wnioskowania SageMaker, a także przesłane przez nas dane szkoleniowe i testowe:

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

Wnioski

Inżynierowie ML, badacze danych i twórcy oprogramowania mogą używać rozwiązania Autopilot do tworzenia i wdrażania potoku wnioskowania przy niewielkim lub żadnym doświadczeniu w programowaniu ML. Autopilot oszczędza czas i zasoby, wykorzystując najlepsze praktyki data science i ML. Duże organizacje mogą teraz przenieść zasoby inżynieryjne z konfiguracji infrastruktury na rzecz ulepszania modeli i rozwiązywania biznesowych przypadków użycia. Startupy i mniejsze organizacje mogą rozpocząć korzystanie z uczenia maszynowego, mając niewielką lub żadną wiedzę z zakresu uczenia maszynowego.

Aby rozpocząć korzystanie z Autopilota SageMaker, zobacz Strona produktu lub uzyskaj dostęp do SageMaker Autopilot w SageMaker Studio.

Zalecamy także zapoznanie się z innymi ważnymi funkcjami, jakie oferuje SageMaker, np Sklep funkcji Amazon SageMaker, który integruje się z Rurociągi Amazon SageMaker do tworzenia, dodawania wyszukiwania i odkrywania funkcji oraz ponownego wykorzystywania zautomatyzowanych przepływów pracy ML. Możesz uruchomić wiele symulacji Autopilota z różnymi wariantami funkcji lub celów w swoim zestawie danych. Można to również potraktować jako problem dynamicznej alokacji pojazdów, w którym model próbuje przewidzieć zapotrzebowanie na pojazdy na podstawie czasu (np. pory dnia lub dnia tygodnia) lub lokalizacji, albo kombinacji obu.

O autorach

Douga Mbaya jest starszym architektem rozwiązań specjalizującym się w danych i analityce. Doug ściśle współpracuje z partnerami AWS, pomagając im integrować rozwiązania danych i analityki w chmurze. Wcześniejsze doświadczenie Douga obejmuje wspieranie klientów AWS w segmencie wspólnych przejazdów i dostaw żywności.

Walerio Perrone jest Applied Science Managerem pracującym nad automatycznym strojeniem modeli Amazon SageMaker i Autopilotem.

• Coinsmart. Najlepsza w Europie giełda bitcoinów i kryptowalut.
• Platoblockchain. Web3 Inteligencja Metaverse. Wzmocniona wiedza. DARMOWY DOSTĘP.
• CryptoJastrząb. Radar Altcoin. Bezpłatna wersja próbna.
• Źródło: https://aws.amazon.com/blogs/machine-learning/automate-a-shared-bikes-and-scooters-classification-model-with-amazon-sagemaker-autopilot/