MLflow로 스파크 확장

안녕하세요, Khabrovites. 우리가 이미 쓴 것처럼 이번 달 OTUS는 기계 학습에 대한 두 가지 과정을 한 번에 시작합니다. 기본 и 고급의. 이와 관련하여 유용한 자료를 계속 공유합니다.

이 기사의 목적은 우리의 첫 번째 경험에 대해 이야기하는 것입니다. ML플로우.

리뷰 시작하겠습니다 ML플로우 추적 서버에서 연구의 모든 반복을 프롤로그합니다. 그런 다음 UDF를 사용하여 Spark와 MLflow를 연결한 경험을 공유합니다.

문맥

우리는 알파 건강 우리는 기계 학습과 인공 지능을 사용하여 사람들이 자신의 건강과 웰빙을 돌볼 수 있도록 지원합니다. 그렇기 때문에 우리가 개발하는 데이터 제품의 핵심에는 머신 러닝 모델이 있고, 머신 러닝 수명 주기의 모든 측면을 포괄하는 오픈 소스 플랫폼인 MLflow가 우리의 주목을 받은 이유입니다.

ML플로우

MLflow의 주요 목표는 데이터 과학자가 거의 모든 기계 학습 라이브러리로 작업할 수 있도록 기계 학습 위에 추가 계층을 제공하는 것입니다.h2o, 케 라스, 도약, 파이 토치, kle 다 и 텐서 흐름), 그녀의 작업을 다음 단계로 끌어 올립니다.

MLflow는 세 가지 구성 요소를 제공합니다.

• 추적 – 실험 기록 및 요청: 코드, 데이터, 구성 및 결과. 모델을 만드는 과정을 따르는 것이 매우 중요합니다.
• 프로젝트 – 모든 플랫폼에서 실행할 수 있는 패키징 형식(예: 세이지 메이커)
• 모델 다양한 배포 도구에 모델을 제출하기 위한 일반적인 형식입니다.

MLflow(작성 당시 알파)는 실험, 재사용 및 배포를 포함하여 기계 학습 수명 주기를 관리할 수 있는 오픈 소스 플랫폼입니다.

MLflow 설정

MLflow를 사용하려면 먼저 전체 Python 환경을 설정해야 합니다. PyEnv (Mac에 Python을 설치하려면 여기에). 따라서 실행에 필요한 모든 라이브러리를 설치할 가상 환경을 만들 수 있습니다.

```
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
```

참고: UDF와 같은 모델을 실행하기 위해 PyArrow를 사용하고 있습니다. 최신 버전이 서로 충돌하여 PyArrow와 Numpy의 버전을 수정해야 했습니다.

추적 UI 실행

MLflow 추적을 통해 Python으로 실험을 기록하고 쿼리할 수 있습니다. REST API. 또한 모델 아티팩트를 저장할 위치를 정의할 수 있습니다(localhost, 아마존 S3, Azure Blob 저장소, Google 클라우드 저장소 또는 SFTP 서버). Alpha Health에서 AWS를 사용하므로 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에서 사용할 수 있습니다.

참고: 추적 UI 및 모델 클라이언트는 아티팩트의 위치에 액세스할 수 있어야 합니다. 즉, 추적 UI가 EC2 인스턴스에 있다는 사실에 관계없이 MLflow를 로컬로 실행할 때 컴퓨터는 아티팩트 모델을 작성하기 위해 S3에 직접 액세스할 수 있어야 합니다.

추적 UI는 S3 버킷에 아티팩트를 저장합니다.

모델 실행

추적 서버가 실행되는 즉시 모델 교육을 시작할 수 있습니다.

예를 들어 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 해시 로그를 사용하여 필요한 반복을 수행한 코드를 이해함으로써 최상의 모델을 재현할 수 있기 때문에 매우 유용합니다.

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 server" 명령으로 시작된 MLflow 추적 서버에는 실행을 추적하고 데이터를 로컬 파일 시스템에 쓰기 위한 REST API가 있습니다. "MLFLOW_TRACKING_URI" 환경 변수를 사용하여 추적 서버의 주소를 지정할 수 있으며 MLflow 추적 API는 자동으로 이 주소에서 추적 서버에 연결하여 실행 정보, 로깅 메트릭 등을 생성/가져옵니다.

출처 : Docs// 추적 서버 실행

모델에 서버를 제공하려면 실행 중인 추적 서버(시작 인터페이스 참조)와 모델의 실행 ID가 필요합니다.

실행 ID

# 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 제공 기능을 사용하여 모델을 제공하려면 간단히 지정하여 모델에 대한 정보를 얻기 위해 추적 UI에 액세스해야 합니다. --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 // 문서 // 모델 #local), Spark(배치 또는 스트리밍)를 사용하는 것은 배포로 인해 훨씬 ​​더 강력한 솔루션입니다.

방금 오프라인 교육을 수행한 다음 출력 모델을 모든 데이터에 적용했다고 상상해 보십시오. 여기에서 Spark와 MLflow가 제 역할을 합니다.

파이스파크 + 주피터 + 스파크 설치

출처 : 시작하기 PySpark - Jupyter

Spark 데이터 프레임에 MLflow 모델을 적용하는 방법을 보여주기 위해 PySpark와 함께 작동하도록 Jupyter 노트북을 설정해야 합니다.

최신 안정 버전을 설치하여 시작하십시오. 아파치 스파크:

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, 원하는 폴더에 노트북을 저장할 수 있습니다.

PySpark에서 Jupyter 실행

Jupiter를 PySpark 드라이버로 설정할 수 있었기 때문에 이제 PySpark 컨텍스트에서 Jupyter 노트북을 실행할 수 있습니다.

(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 - 와인 품질 예측

지금까지 전체 와인 데이터 세트에서 와인 품질 예측을 실행하여 MLflow와 함께 PySpark를 사용하는 방법에 대해 이야기했습니다. 하지만 Scala Spark에서 Python MLflow 모듈을 사용해야 하는 경우에는 어떻게 해야 할까요?

Scala와 Python 간에 Spark 컨텍스트를 분할하여 이를 테스트했습니다. 즉, Python에 MLflow UDF를 등록하고 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 파이프라인 및 추천 시스템에 사용할 것입니다.

파일 시스템 대신 데이터베이스와 파일 스토리지를 동기화하는 것이 좋을 것입니다. 이렇게 하면 동일한 파일 공유를 사용할 수 있는 여러 엔드포인트가 제공됩니다. 예를 들어 여러 인스턴스를 사용합니다. 프레스토 악장 и 아테나 동일한 Glue metastore로.

요약하면 데이터 작업을 더욱 흥미롭게 만들어준 MLFlow 커뮤니티에 감사의 말을 전하고 싶습니다.

MLflow를 사용하는 경우 언제든지 저희에게 편지를 보내 사용 방법을 알려주십시오. 프로덕션에서 사용하는 경우에는 더욱 그렇습니다.

과정에 대해 자세히 알아보기:
기계 학습. 기본 과정
기계 학습. 고급 코스

더 읽어보기:

• 감독 학습 문제에 주성분 분석을 적용할 때의 위험 및 주의 사항
• Docker로 기계 학습 모델 배포 - 1부
• Docker로 기계 학습 모델 배포 - 2부

출처 : habr.com