🥇הרחבת ספארק עם MLflow

שלום תושבי חברובסק. כפי שכבר כתבנו, החודש OTUS משיקה שני קורסי למידת מכונה בבת אחת, כלומר базовый и מִתקַדֵם. בהקשר זה, אנו ממשיכים לחלוק חומר שימושי.

מטרת מאמר זה היא לדבר על הניסיון הראשון שלנו בשימוש זרימת ML.

נתחיל בסקירה זרימת ML משרת המעקב שלו ורשום את כל האיטרציות של המחקר. לאחר מכן נשתף את החוויה שלנו בחיבור Spark עם MLflow באמצעות UDF.

הֶקשֵׁר

אנחנו בפנים אלפא בריאות אנו משתמשים בלמידת מכונה ובינה מלאכותית כדי להעצים אנשים לקחת אחריות על בריאותם ורווחתם. זו הסיבה שמודלים של למידת מכונה נמצאים בלב מוצרי מדעי הנתונים שאנו מפתחים, וזו הסיבה שנמשכנו ל-MLflow, פלטפורמת קוד פתוח המכסה את כל ההיבטים של מחזור החיים של למידת מכונה.

זרימת ML

המטרה העיקרית של MLflow היא לספק שכבה נוספת על גבי למידת מכונה שתאפשר למדעני נתונים לעבוד כמעט עם כל ספריית למידת מכונה (h2o, keras, קפיצת מדרגה, פיטורך, למד и tensorflow), לוקח את עבודתה לשלב הבא.

MLflow מספק שלושה מרכיבים:

מעקב - הקלטה ובקשות לניסויים: קוד, נתונים, תצורה ותוצאות. מעקב אחר תהליך יצירת המודל חשוב מאוד.
פרויקטים - פורמט אריזה להפעלה בכל פלטפורמה (למשל. SageMaker)
מודלים – פורמט נפוץ להגשת מודלים לכלי פריסה שונים.

MLflow (באלפא בזמן הכתיבה) היא פלטפורמת קוד פתוח המאפשרת לך לנהל את מחזור החיים של למידת מכונה, כולל ניסויים, שימוש חוזר ופריסה.

הגדרת MLflow

כדי להשתמש ב-MLflow תחילה עליך להגדיר את כל סביבת Python שלך, לשם כך נשתמש PyEnv (כדי להתקין Python ב-Mac, בדוק כאן). כך נוכל ליצור סביבה וירטואלית שבה נתקין את כל הספריות הנחוצות להפעלתה.

```
pyenv install 3.7.0
pyenv global 3.7.0 # Use Python 3.7
mkvirtualenv mlflow # Create a Virtual Env with Python 3.7
workon mlflow
```

בואו נתקין את הספריות הנדרשות.

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

הערה: אנו משתמשים ב- PyArrow כדי להפעיל מודלים כגון UDF. היה צורך לתקן את הגירסאות של PyArrow ו-Numpy מכיוון שהגירסאות האחרונות התנגשו זו בזו.

הפעל את ממשק המשתמש למעקב

מעקב MLflow מאפשר לנו לרשום ולשאול ניסויים באמצעות Python ו REST ממשק API. בנוסף, אתה יכול לקבוע היכן לאחסן חפצי מודל (localhost, אמזון S3, אחסון Azure Blob, אחסון בענן של Google או שרת SFTP). מכיוון שאנו משתמשים ב-AWS ב-Alpha Health, אחסון החפצים שלנו יהיה S3.

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

MLflow ממליצה להשתמש באחסון קבצים מתמשך. אחסון קבצים הוא המקום שבו השרת יאחסן מטא נתונים של ריצה וניסוי. בעת הפעלת השרת, ודא שהוא מצביע על מאגר הקבצים הקבוע. כאן עבור הניסוי פשוט נשתמש /tmp.

זכור שאם ברצוננו להשתמש בשרת mlflow כדי להפעיל ניסויים ישנים, הם חייבים להיות נוכחים באחסון הקבצים. עם זאת, גם בלי זה נוכל להשתמש בהם ב-UDF, מכיוון שאנו צריכים רק את הנתיב למודל.

הערה: זכור שממשק המשתמש למעקב וללקוח הדגם חייבת להיות בעלת גישה למיקום החפץ. כלומר, ללא קשר לעובדה שממשק המשתמש למעקב נמצא במופע EC2, בעת הפעלת MLflow באופן מקומי, למכונה חייבת להיות גישה ישירה ל-S3 כדי לכתוב מודלים של חפצים.

ממשק המשתמש למעקב מאחסן חפצים בדלי S3

דגמים רצים

ברגע ששרת ה-Tracking פועל, אתה יכול להתחיל לאמן את הדגמים.

כדוגמה, נשתמש בשינוי היין מדוגמה של MLflow ב סקלרן.

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

כפי שכבר דיברנו, MLflow מאפשר לך לרשום פרמטרים, מדדים וחפצים של מודל כדי שתוכל לעקוב אחר האופן שבו הם מתפתחים במהלך איטרציות. תכונה זו שימושית ביותר מכיוון שכך נוכל לשחזר את המודל הטוב ביותר על ידי פנייה לשרת המעקב או להבין איזה קוד ביצע את האיטרציה הנדרשת באמצעות יומני ה-git hash של commits.

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

איטרציות יין

חלק שרת לדגם

לשרת המעקב MLflow, המופעל באמצעות הפקודה "mlflow server", יש REST API למעקב אחר ריצות וכתיבת נתונים למערכת הקבצים המקומית. אתה יכול לציין את כתובת שרת המעקב באמצעות משתנה הסביבה "MLFLOW_TRACKING_URI" וה-API למעקב של MLflow ייצור קשר אוטומטית עם שרת המעקב בכתובת זו כדי ליצור/לקבל מידע על השקה, מדדי יומן וכו'.

מקור: Docs// הפעלת שרת מעקב

כדי לספק למודל שרת, אנו זקוקים לשרת מעקב פועל (ראה ממשק השקה) ואת מזהה ההפעלה של המודל.

מזהה הפעלה

# 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

כדי לשרת מודלים המשתמשים בפונקציונליות השירות של MLflow, נצטרך גישה לממשק המשתמש למעקב כדי לקבל מידע על המודל פשוט על ידי ציון --run_id.

ברגע שהדגם יוצר קשר עם שרת המעקב, נוכל לקבל נקודת קצה חדשה של הדגם.

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

ריצה דגמים מבית Spark

למרות העובדה ששרת המעקב חזק מספיק כדי לשרת דגמים בזמן אמת, יש לאמן אותם ולהשתמש בפונקציונליות השרת (מקור: mlflow // docs // models # local), השימוש ב-Spark (אצווה או סטרימינג) הוא פתרון חזק עוד יותר בשל ההפצה.

תארו לעצמכם שפשוט ביצעתם את ההדרכה במצב לא מקוון ואז החלתם את מודל הפלט על כל הנתונים שלכם. זה המקום שבו Spark ו-MLflow זוהרים.

התקן PySpark + Jupyter + Spark

מקור: התחל PySpark - Jupyter

כדי להראות כיצד אנו מיישמים מודלים של MLflow על מסגרות נתונים של Spark, עלינו להגדיר מחברות Jupyter כך שיעבדו יחד עם PySpark.

התחל בהתקנת הגרסה היציבה האחרונה אפאצ 'י ספארק:

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̀

התקן את PySpark ו-Jupyter בסביבה הוירטואלית:

pip install pyspark jupyter

הגדר משתני סביבה:

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"

לאחר שנקבע notebook-dir, נוכל לאחסן את המחברות שלנו בתיקייה הרצויה.

השקת Jupyter מבית PySpark

מכיוון שהצלחנו להגדיר את Jupiter כמנהל התקן PySpark, אנו יכולים כעת להפעיל את Jupyter Notebook בהקשר של PySpark.

(mlflow) afranzi:~$ pyspark
[I 19:05:01.572 NotebookApp] sparkmagic extension enabled!
[I 19:05:01.573 NotebookApp] Serving notebooks from local directory: /Users/afranzi/Projects/notebooks
[I 19:05:01.573 NotebookApp] The Jupyter Notebook is running at:
[I 19:05:01.573 NotebookApp] http://localhost:8888/?token=c06252daa6a12cfdd33c1d2e96c8d3b19d90e9f6fc171745
[I 19:05:01.573 NotebookApp] Use Control-C to stop this server and shut down all kernels (twice to skip confirmation).
[C 19:05:01.574 NotebookApp]

    Copy/paste this URL into your browser when you connect for the first time,
    to login with a token:
        http://localhost:8888/?token=c06252daa6a12cfdd33c1d2e96c8d3b19d90e9f6fc171745

כפי שהוזכר לעיל, MLflow מספק תכונה לרישום חפצי מודל ב-S3. ברגע שיש לנו את הדגם הנבחר בידינו, יש לנו הזדמנות לייבא אותו כ-UDF באמצעות המודול mlflow.pyfunc.

import mlflow.pyfunc

model_path = 's3://<bucket>/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 - פלט תחזיות איכות היין

עד לנקודה זו, דיברנו על איך להשתמש ב-PySpark עם MLflow, הפעלת תחזיות איכות יין על כל מערך הנתונים של היין. אבל מה אם אתה צריך להשתמש במודולי Python MLflow מ- Scala Spark?

בדקנו גם את זה על ידי פיצול ההקשר של Spark בין Scala לפייתון. כלומר, רשמנו את MLflow UDF ב-Python, והשתמשנו בו מ-Scala (כן, אולי לא הפתרון הכי טוב, אבל מה שיש לנו).

סקאלה ספארק + MLflow

לדוגמא זו נוסיף טורי גרעין לתוך צדק הקיים.

התקן את Spark + Toree + Jupyter

pip install toree
jupyter toree install --spark_home=${SPARK_HOME} --sys-prefix
jupyter kernelspec list
```
```
Available kernels:
  apache_toree_scala    /Users/afranzi/.virtualenvs/mlflow/share/jupyter/kernels/apache_toree_scala
  python3               /Users/afranzi/.virtualenvs/mlflow/share/jupyter/kernels/python3
```

כפי שניתן לראות מהמחברת המצורפת, ה-UDF משותף בין Spark ל-PySpark. אנו מקווים שחלק זה יהיה שימושי עבור אלה שאוהבים את Scala ורוצים לפרוס מודלים של למידת מכונה בייצור.

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.<locals>.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       |
+-----------+--------+-----------+---------+-----------+

השלבים הבאים

למרות ש-MLflow נמצא בגרסת אלפא בזמן הכתיבה, זה נראה מבטיח למדי. רק היכולת להפעיל מסגרות למידת מכונה מרובות ולצרוך אותן מנקודת קצה אחת מעלה את מערכות הממליצים לשלב הבא.

בנוסף, MLflow מקרבת את מהנדסי הנתונים ומומחי מדעי הנתונים, ומניחת שכבה משותפת ביניהם.

לאחר חקירה זו של MLflow, אנו בטוחים שנתקדם ונשתמש בו עבור צינורות Spark ומערכות הממליצים שלנו.

זה יהיה נחמד לסנכרן את אחסון הקבצים עם מסד הנתונים במקום עם מערכת הקבצים. זה אמור לתת לנו מספר נקודות קצה שיכולות להשתמש באותו אחסון קבצים. לדוגמה, השתמש במספר מופעים פרסטו и אתנה עם אותו Metastore של Glue.

לסיכום, אני רוצה לומר תודה לקהילת MLFlow על שהפכת את העבודה שלנו עם נתונים למעניינת יותר.

אם אתה משחק עם MLflow, אל תהסס לכתוב לנו ולספר לנו איך אתה משתמש בו, ועוד יותר אם אתה משתמש בו בהפקה.

למידע נוסף על הקורסים:
למידת מכונה. קורס בסיסי
למידת מכונה. קורס מתקדם

קרא עוד:

מקור: www.habr.com

הרחבת Spark עם MLflow