Rozszerzenie możliwości Spark za pomocą MLflow

Cześć, mieszkańcy Habr. Jak już pisaliśmy, w tym miesiącu OTUS uruchamia dwa kursy dotyczące uczenia maszynowego, a mianowicie podstawowy i zaawansowany. W związku z tym kontynuujemy dzielenie się przydatnymi materiałami.

Celem tego artykułu jest opowiedzenie o naszym pierwszym doświadczeniu w korzystaniu z MLflow.

. Rozpoczniemy przegląd MLflow od jego serwera śledzenia i zarejestrujemy wszystkie iteracje badania. Następnie podzielimy się doświadczeniem łączenia Sparka z MLflow za pomocą UDF.

Kontekst

My w Alpha Health wykorzystujemy uczenie maszynowe i sztuczną inteligencję, aby umożliwić ludziom dbałość o ich zdrowie i dobrostan. Dlatego modele uczenia maszynowego leżą u podstaw opracowywanych przez nas produktów do przetwarzania danych, a właśnie dlatego zwróciliśmy uwagę na MLflow — platformę z otwartym kodem źródłowym, która obejmuje wszystkie aspekty cyklu życia uczenia maszynowego.

MLflow

Głównym celem MLflow jest zapewnienie dodatkowej warstwy nad uczeniem maszynowym, która pozwoliłaby specjalistom ds. danych pracować praktycznie z każdą biblioteką uczenia maszynowego (h2o, keras, mleap, pytorch, sklearn i tensorflow), podnosząc jej działanie na wyższy poziom.

MLflow zapewnia trzy komponenty:

• Śledzenie – rejestracja i zapytania o eksperymenty: kod, dane, konfiguracja i wyniki. Monitorowanie procesu tworzenia modelu jest bardzo ważne.
• Projekty – Format pakowania do uruchomienia na dowolnej platformie (np. SageMaker)
• Modele – ogólny format wdrażania modeli w różne narzędzia.

MLflow (w momencie pisania artykułu w wersji alfa) — platforma z otwartym kodem źródłowym, która pozwala zarządzać cyklem życia uczenia maszynowego, w tym eksperymentami, ponownym użyciem i wdrożeniem.

Konfiguracja MLflow

Aby korzystać z MLflow, najpierw musisz skonfigurować całe środowisko Python, w tym celu skorzystamy z PyEnv (aby zainstalować Python na Mac, zapoznaj się z tutaj). W ten sposób będziemy mogli utworzyć wirtualne środowisko, do którego zainstalujemy wszystkie niezbędne biblioteki.

```
pyenv install 3.7.0
pyenv global 3.7.0 # Użyj Pythona 3.7
mkvirtualenv mlflow # Utwórz wirtualne środowisko z Pythonem 3.7
workon mlflow
```

Zainstalujemy wymagane biblioteki.

```
pip install mlflow==0.7.0 
            Cython==0.29  
            numpy==1.14.5 
            pandas==0.23.4 
            pyarrow==0.11.0
```

Uwaga: używamy PyArrow do uruchamiania modeli takich jak UDF. Wersje PyArrow i Numpy musiały być dostosowane, ponieważ najnowsze wersje były ze sobą niekompatybilne.

Uruchamiamy interfejs śledzenia

MLflow Tracking umożliwia nam logowanie oraz zapytania do eksperymentów za pomocą Pythona i REST API. Ponadto możemy określić, gdzie przechowywać artefakty modelu (localhost, Amazon S3, Azure Blob Storage, Google Cloud Storage lub serwer SFTP). Ponieważ w Alpha Health korzystamy z AWS, jako magazyn artefaktów będzie S3.

# Running a Tracking Server
mlflow server 
    --file-store /tmp/mlflow/fileStore 
    --default-artifact-root s3://<bucket>/mlflow/artifacts/ 
    --host localhost
    --port 5000

MLflow zaleca korzystanie z trwałego magazynu plików. Magazyn plików to miejsce, w którym serwer przechowuje metadane uruchomień i eksperymentów. Podczas uruchamiania serwera upewnij się, że wskazuje on na trwały magazyn plików. Tutaj w eksperymencie po prostu użyjemy /tmp.

Pamiętaj, że jeśli chcemy używać serwera MLflow do uruchamiania starych eksperymentów, muszą one być obecne w magazynie plików. Jednak nawet bez tego moglibyśmy je wykorzystać w UDF, ponieważ potrzebujemy tylko ścieżki do modelu.

Uwaga: Pamiętaj, że Tracking UI i klient modelu muszą mieć dostęp do lokalizacji artefaktu. To znaczy, niezależnie od tego, że Tracking UI jest w instancji EC2, przy lokalnym uruchamianiu MLflow maszyna musi mieć bezpośredni dostęp do S3, aby zapisywać modele artefaktów.

Tracking UI przechowuje artefakty w kuble S3

Uruchamianie modeli

Gdy serwer Tracking będzie działać, można rozpocząć trenowanie modeli.

Na przykład skorzystamy z modyfikacji wine z przykładu MLflow w Sklearn.

MLFLOW_TRACKING_URI=http://localhost:5000 python wine_quality.py 
  --alpha 0.9
  --l1_ration 0.5
  --wine_file ./data/winequality-red.csv

Jak już wspomniano, MLflow umożliwia logowanie parametrów, metryk i artefaktów modeli, aby można było śledzić, jak się rozwijają na przestrzeni iteracji. Ta funkcja jest niezwykle użyteczna, ponieważ pozwala nam odtworzyć najlepszy model, odwołując się do serwera Tracking lub rozumiejąc, jaki kod wykonał potrzebną iterację, korzystając z logów hash commitów git.

with mlflow.start_run():

    ... model ...

    mlflow.log_param("source", wine_path)
    mlflow.log_param("alpha", alpha)
    mlflow.log_param("l1_ratio", l1_ratio)

    mlflow.log_metric("rmse", rmse)
    mlflow.log_metric("r2", r2)
    mlflow.log_metric("mae", mae)

    mlflow.set_tag('domain', 'wine')
    mlflow.set_tag('predict', 'quality')
    mlflow.sklearn.log_model(lr, "model")

Iteracje wine

Backend dla modelu

Serwer śledzenia MLflow uruchomiony za pomocą polecenia „mlflow server” ma interfejs REST API do śledzenia uruchomień i zapisywania danych w lokalnym systemie plików. Możesz określić adres serwera śledzenia za pomocą zmiennej środowiskowej „MLFLOW_TRACKING_URI”, a API śledzenia MLflow automatycznie połączy się z serwerem śledzenia pod tym adresem, aby utworzyć/otrzymać informacje o uruchomieniu, metryki logów itd.

Źródło: Dokumentacja// Uruchamianie serwera śledzenia

Aby zapewnić modelowi serwer, potrzebujemy uruchomionego serwera śledzenia (zob. interfejs uruchamiania) oraz identyfikatora uruchomienia modelu.

Identyfikator uruchomienia

# Serve a sklearn model through 127.0.0.0:5005
MLFLOW_TRACKING_URI=http://0.0.0.0:5000 mlflow sklearn serve 
  --port 5005  
  --run_id 0f8691808e914d1087cf097a08730f17 
  --model-path model

Aby obsługiwać modele za pomocą funkcji MLflow serve, potrzebujemy dostępu do interfejsu użytkownika śledzenia, aby uzyskać informacje o modelu, po prostu podając --run_id.

Gdy model połączy się z serwerem śledzenia, możemy uzyskać nowy punkt końcowy modelu.

# Query Tracking Server Endpoint
curl -X POST 
  http://127.0.0.1:5005/invocations 
  -H 'Content-Type: application/json' 
  -d '[
	{
		"fixed acidity": 3.42, 
		"volatile acidity": 1.66, 
		"citric acid": 0.48, 
		"residual sugar": 4.2, 
		"chloridessssss": 0.229, 
		"free sulfur dsioxide": 19, 
		"total sulfur dioxide": 25, 
		"density": 1.98, 
		"pH": 5.33, 
		"sulphates": 4.39, 
		"alcohol": 10.8
	}
]'

> {"predictions": [5.825055635303461]}

Uruchamianie modeli ze Spark

Chociaż serwer śledzenia jest wystarczająco potężny, aby obsługiwać modele w czasie rzeczywistym, ich szkolenie i wykorzystanie funkcji serve (źródło: mlflow // docs // models # local), wykorzystanie Sparka (wsadowo lub strumieniowo) – to jeszcze potężniejsze rozwiązanie dzięki rozproszonemu przetwarzaniu.

Wyobraź sobie, że właśnie zakończyłeś szkolenie offline, a następnie zastosowałeś model wyjściowy do wszystkich swoich danych. Właśnie w tym miejscu Spark i MLflow pokażą swoją najlepszą stronę.

Instalujemy PySpark + Jupyter + Spark

Źródło: Rozpocznij od PySpark — Jupyter

Aby pokazać, jak stosujemy modele MLflow do ram danych Sparka, musimy skonfigurować współpracę notatników Jupyter z PySpark.

Zacznij od zainstalowania najnowszej stabilnej wersji Apache Spark:

cd ~/Downloads/
tar -xzf spark-2.4.3-bin-hadoop2.7.tgz
mv ~/Downloads/spark-2.4.3-bin-hadoop2.7 ~/
ln -s ~/spark-2.4.3-bin-hadoop2.7 ~/spark

Zainstaluj PySpark i Jupyter w wirtualnym środowisku:

pip install pyspark jupyter

Skonfiguruj zmienne środowiskowe:

export SPARK_HOME=~/spark
export PATH=$SPARK_HOME/bin:$PATH
export PYSPARK_DRIVER_PYTHON=jupyter
export PYSPARK_DRIVER_PYTHON_OPTS="notebook --notebook-dir=${HOME}/Projects/notebooks"

Definiując notebook-dir, będziemy mogli przechowywać nasze notatniki w wybranym folderze.

Uruchamiamy Jupyter z PySpark

Ponieważ udało nam się skonfigurować Jupyter jako sterownik PySpark, możemy teraz uruchomić notatnik Jupyter w kontekście PySpark.

(mlflow) afranzi:~$ pyspark
[I 19:05:01.572 NotebookApp] Włączono rozszerzenie sparkmagic!
[I 19:05:01.573 NotebookApp] Serweruje notatniki z lokalnego katalogu: /Users/afranzi/Projects/notebooks
[I 19:05:01.573 NotebookApp] Jupyter Notebook działa pod adresem:
[I 19:05:01.573 NotebookApp] http://localhost:8888/?token=c06252daa6a12cfdd33c1d2e96c8d3b19d90e9f6fc171745
[I 19:05:01.573 NotebookApp] Użyj Control-C, aby zatrzymać ten serwer i zamknąć wszystkie rdzenie (naciśnij dwa razy, aby pominąć potwierdzenie).
[C 19:05:01.574 NotebookApp]

    Skopiuj/wklej ten adres URL do przeglądarki podczas pierwszego połączenia,
    aby zalogować się z użyciem tokena:
        http://localhost:8888/?token=c06252daa6a12cfdd33c1d2e96c8d3b19d90e9f6fc171745

Jak wspomniano wcześniej, MLflow oferuje funkcję logowania artefaktów modelu do S3. Gdy już mamy w rękach wybrany model, możemy zaimportować go jako UDF za pomocą modułu mlflow.pyfunc.

import mlflow.pyfunc

model_path = 's3:///mlflow/artifacts/1/0f8691808e914d1087cf097a08730f17/artifacts/model'
wine_path = '/Users/afranzi/Projects/data/winequality-red.csv'
wine_udf = mlflow.pyfunc.spark_udf(spark, model_path)

df = spark.read.format("csv").option("header", "true").option('delimiter', ';').load(wine_path)
columns = [ "fixed acidity", "volatile acidity", "citric acid",
            "residual sugar", "chlorides", "free sulfur dioxide",
            "total sulfur dioxide", "density", "pH",
            "sulphates", "alcohol"
          ]
          
df.withColumn('prediction', wine_udf(*columns)).show(100, False)

PySpark – Prognoza jakości wina

Do tej pory omawialiśmy, jak używać PySpark z MLflow, uruchamiając prognozowanie jakości wina na całym zbiorze danych wine. Ale co zrobić, jeśli trzeba użyć modułów Python MLflow z Scala Spark?

Przetestowaliśmy to, dzieląc kontekst Spark między Scalę a Pythona. Oznacza to, że zarejestrowaliśmy MLflow UDF w Pythonie i wykorzystaliśmy go z SCala (tak, może to nie najlepsze rozwiązanie, ale co mamy).

Scala Spark + MLflow

Dla tego przykładu dodamy Toree Kernel do istniejącego Jupiter.

Instalujemy Spark + Toree + Jupyter

pip install toree
jupyter toree install --spark_home=${SPARK_HOME} --sys-prefix
jupyter kernelspec list
```
```
Dostępne jądra:
  apache_toree_scala    /Users/afranzi/.virtualenvs/mlflow/share/jupyter/kernels/apache_toree_scala
  python3               /Users/afranzi/.virtualenvs/mlflow/share/jupyter/kernels/python3
```

Jak widać z dołączonego notatnika, UDF jest używane razem ze Spark i PySpark. Mamy nadzieję, że ta część będzie przydatna dla tych, którzy lubią Scalę i chcą wdrożyć modele uczenia maszynowego na produkcji.

import org.apache.spark.sql.functions.col
import org.apache.spark.sql.types.StructType
import org.apache.spark.sql.{Column, DataFrame}
import scala.util.matching.Regex

val FirstAtRe: Regex = "^_".r
val AliasRe: Regex = "[\s_.:@]+".r

def getFieldAlias(field_name: String): String = {
    FirstAtRe.replaceAllIn(AliasRe.replaceAllIn(field_name, "_"), "")
}

def selectFieldsNormalized(columns: List[String])(df: DataFrame): DataFrame = {
    val fieldsToSelect: List[Column] = columns.map(field =>
        col(field).as(getFieldAlias(field))
    )
    df.select(fieldsToSelect: _*)
}

def normalizeSchema(df: DataFrame): DataFrame = {
    val schema = df.columns.toList
    df.transform(selectFieldsNormalized(schema))
}

FirstAtRe = ^_
AliasRe = [s_.:@]+

getFieldAlias: (field_name: String)String
selectFieldsNormalized: (columns: List[String])(df: org.apache.spark.sql.DataFrame)org.apache.spark.sql.DataFrame
normalizeSchema: (df: org.apache.spark.sql.DataFrame)org.apache.spark.sql.DataFrame
Out[1]:
[s_.:@]+
In [2]:
val winePath = "~\/Research\/mlflow-workshop\/examples\/wine_quality\/data\/winequality-red.csv"
val modelPath = "\/tmp\/mlflow\/artifactStore\/0\/96cba14c6e4b452e937eb5072467bf79\/artifacts\/model"

winePath = ~\/Research\/mlflow-workshop\/examples\/wine_quality\/data\/winequality-red.csv
modelPath = \/tmp\/mlflow\/artifactStore\/0\/96cba14c6e4b452e937eb5072467bf79\/artifacts\/model
Out[2]:
\/tmp\/mlflow\/artifactStore\/0\/96cba14c6e4b452e937eb5072467bf79\/artifacts\/model
In [3]:
val df = spark.read
              .format("csv")
              .option("header", "true")
              .option("delimiter", ";")
              .load(winePath)
              .transform(normalizeSchema)

df = [fixed_acidity: string, volatile_acidity: string ... 10 more fields]
Out[3]:
[fixed_acidity: string, volatile_acidity: string ... 10 more fields]
In [4]:
%%PySpark
import mlflow
from mlflow import pyfunc

model_path = "\/tmp\/mlflow\/artifactStore\/0\/96cba14c6e4b452e937eb5072467bf79\/artifacts\/model"
wine_quality_udf = mlflow.pyfunc.spark_udf(spark, model_path)

spark.udf.register("wineQuality", wine_quality_udf)
Out[4]:
<function spark_udf..predict at 0x1116a98c8>
In [6]:
df.createOrReplaceTempView("wines")
In [10]:
%%SQL
SELECT 
    quality,
    wineQuality(
        fixed_acidity,
        volatile_acidity,
        citric_acid,
        residual_sugar,
        chlorides,
        free_sulfur_dioxide,
        total_sulfur_dioxide,
        density,
        pH,
        sulphates,
        alcohol
    ) AS prediction
FROM wines
LIMIT 10
Out[10]:
+-------+------------------+
|quality|        prediction|
+-------+------------------+
|      5| 5.576883967129615|
|      5|  5.50664776916154|
|      5| 5.525504822954496|
|      6| 5.504311247097457|
|      5| 5.576883967129615|
|      5|5.5556903912725755|
|      5| 5.467882654744997|
|      7| 5.710602976324739|
|      7| 5.657319539336507|
|      5| 5.345098606538708|
+-------+------------------+

In [17]:
spark.catalog.listFunctions.filter('name like "%wineQuality%").show(20, false)

+-----------+--------+-----------+---------+-----------+
|name       |database|description|className|isTemporary|
+-----------+--------+-----------+---------+-----------+
|wineQuality|null    |null       |null     |true       |
+-----------+--------+-----------+---------+-----------+

Następne kroki

Mimo że w momencie pisania tego artykułu MLflow jest w wersji Alpha, wygląda obiecująco. Samo możliwość uruchamiania wielu frameworków uczenia maszynowego i korzystania z nich z jednego punktu końcowego podnosi systemy rekomendacyjne na nowy poziom.

Co więcej, MLflow łączy inżynierów danych i specjalistów ds. nauki o danych, tworząc wspólną warstwę między nimi.

Po tym badaniu MLflow jesteśmy pewni, że będziemy kontynuować i używać go do naszych pipeline'ów Sparka oraz w systemach rekomendacyjnych.

Byłoby dobrze zsynchronizować magazyn plików z bazą danych, zamiast z systemem plików. Dzięki temu powinniśmy uzyskać kilka punktów końcowych, które mogą korzystać z tego samego magazynu plików. Na przykład, użyć kilku instancji Presto i Athena z tym samym metastore Glue.

Podsumowując, chciałbym podziękować społeczności MLFlow za to, że czynicie naszą pracę z danymi bardziej interesującą.

Jeśli bawicie się MLflow, nie krępujcie się pisać do nas i opowiadać, jak go używacie, zwłaszcza jeśli wykorzystujecie go w produkcji.

Dowiedz się więcej o kursach:
Machine Learning. Kurs podstawowy
Machine Learning. Kurs zaawansowany

Czytaj dalej:

• Ryzyka i ostrzeżenia przy stosowaniu metody głównych komponentów w zadaniach uczenia nadzorowanego
• Wdrażanie modelu uczenia maszynowego z Dockerem – Część 1
• Wdrażanie modelu uczenia maszynowego z Dockerem – Część 2

Źródło: habr.com