<jupyter_start><jupyter_text>Zajęcia 3 Podsumowanie poprzednich zajęć:* Zapoznanie się z dwoma środowiskami obliczeniowymi (klaster oraz chmura Google). * Operacje na rozproszonym systemie plików (HDFS) i obiektowym systemie plików (GCS)* Analiza rozproszonych danych przy wykorzystaniu Apache Spark Core (API DataFrame), Spark SQL.* Dostęp do danych poprzez biblioteke pandaa, wizualizacja danych (matplotlib/seaborn)* Tworzenie modelu klasyfikacji danych  Zakres dzisiejszych zajęć:* egzamin * wdrożenie modelu ML Zarządzanie i wdrażanie modeli MLPlatform ML Flow (https://www.mlflow.org/docs/latest/index.html) umożliwia całościowe zarządzanie cyklem życia modeli.* logowanie eksperymentów, wartości parametrów modeli i osiąganych przez nie wyników* serializowanie modeli (na potrzeby współdzielenia modelu, przeniesienia na inne środowisko lub serwowania)* wersjonowanie modelu, adnotowanie i przechowywanie w Rejestrze  Przygotowanie danych do analizy<jupyter_code>import os
user_name = os.environ.get('USER')

from pyspark.sql import SparkSession
spark = SparkSession \
.builder \
.config('spark.driver.memory','1g') \
.config('spark.executor.memory', '2g') \
.getOrCreate()

gs_path = f'gs://bdg-lab-{user_name}/survey/2020/survey_results_public.csv'
db_name = user_name.replace('-','_')
spark.sql(f'DROP DATABASE IF EXISTS {db_name} CASCADE')
spark.sql(f'CREATE DATABASE {db_name}')
spark.sql(f'USE {db_name}')
table_name = "survey_2020" 

spark.sql(f'DROP TABLE IF EXISTS {table_name}')

spark.sql(f'CREATE TABLE IF NOT EXISTS {table_name} \
          USING csv \
          OPTIONS (HEADER true, INFERSCHEMA true, NULLVALUE "NA") \
          LOCATION "{gs_path}"')

spark_df= spark.sql(f'SELECT *, CAST((convertedComp > 60000) AS STRING) AS compAboveAvg \
                    FROM {table_name} WHERE convertedComp IS NOT NULL ')<jupyter_output><empty_output><jupyter_text>Transformacja danych do wektora cech<jupyter_code>from pyspark.ml.feature import StringIndexer, OneHotEncoder, VectorAssembler
from pyspark.ml import Pipeline
y = 'compAboveAvg' 
feature_columns = ['OpSys', 'EdLevel', 'MainBranch' , 'Country', 'JobSeek', 'YearsCode']

stringindexer_stages = [StringIndexer(inputCol=c, outputCol='strindexed_' + c).setHandleInvalid("keep") for c in feature_columns]
stringindexer_stages += [StringIndexer(inputCol=y, outputCol='label').setHandleInvalid("keep")]

onehotencoder_stages = [OneHotEncoder(inputCol='strindexed_' + c, outputCol='onehot_' + c) for c in feature_columns]
extracted_columns = ['onehot_' + c for c in feature_columns]
vectorassembler_stage = VectorAssembler(inputCols=extracted_columns, outputCol='features') 

final_columns = [y] + feature_columns + extracted_columns + ['features', 'label']

transformed_df = Pipeline(stages=stringindexer_stages + \
                          onehotencoder_stages + \
                          [vectorassembler_stage]).fit(spark_df).transform(spark_df).select(final_columns)
training, test = transformed_df.randomSplit([0.8, 0.2], seed=1234) # Podzial na zbior treningowy/testowy<jupyter_output><empty_output><jupyter_text>ML Flow: Definicja eksperymentu<jupyter_code>import mlflow
import mlflow.spark

ename = f"classifier_{user_name}"
mlflow.set_experiment(experiment_name=ename)
experiment = mlflow.get_experiment_by_name(ename)<jupyter_output><empty_output><jupyter_text>Definicja metryk<jupyter_code>from pyspark.ml.evaluation import BinaryClassificationEvaluator, MulticlassClassificationEvaluator
evaluator_auroc = BinaryClassificationEvaluator(rawPredictionCol="rawPrediction", metricName="areaUnderROC")
evaluator_acc = MulticlassClassificationEvaluator(labelCol="label", predictionCol="prediction", metricName="accuracy")
evaluator_recall = MulticlassClassificationEvaluator(labelCol="label", predictionCol="prediction", metricName="weightedRecall")
evaluator_prec = MulticlassClassificationEvaluator(labelCol="label", predictionCol="prediction", metricName="weightedPrecision")
evaluator_f = MulticlassClassificationEvaluator(labelCol="label", predictionCol="prediction", metricName="weightedFMeasure")<jupyter_output><empty_output><jupyter_text>ML Flow: logowanie eksperymentu z drzewem decyzyjnym<jupyter_code>from pyspark.ml.classification import DecisionTreeClassifier

dt = DecisionTreeClassifier(featuresCol='features', labelCol='label')
dt_model = Pipeline(stages=[dt]).fit(training)
pred_dt = dt_model.transform(test)
label_and_pred = pred_dt.select('label', 'prediction')

with mlflow.start_run(experiment_id = experiment.experiment_id, run_name="dt_model"):
  
    mlflow.log_param("depth", dt.getMaxDepth())
    res = dt_model.transform(test)

    test_metric_auroc = evaluator_auroc.evaluate(res)
    test_metric_acc = evaluator_acc.evaluate(res)
    test_metric_recall = evaluator_recall.evaluate(res)
    test_metric_prec = evaluator_prec.evaluate(res)
    test_metric_f = evaluator_f.evaluate(res)

    mlflow.log_metric(evaluator_auroc.getMetricName(), test_metric_auroc) 
    mlflow.log_metric(evaluator_acc.getMetricName(), test_metric_acc) 
    mlflow.log_metric(evaluator_recall.getMetricName(), test_metric_recall) 
    mlflow.log_metric(evaluator_prec.getMetricName(), test_metric_prec)     
    mlflow.log_metric(evaluator_f.getMetricName(), test_metric_f) 
  
    mlflow.spark.log_model(spark_model=dt_model, artifact_path='dt_classifier')<jupyter_output><empty_output><jupyter_text>ML Flow: logowanie eksperymentu drzewa decyzyjnego z walidacją krzyżową<jupyter_code>from pyspark.ml.tuning import ParamGridBuilder
param_grid = ParamGridBuilder().\
    addGrid(dt.maxDepth, [2,3,4,5,6]).\
    build()
from pyspark.ml.tuning import CrossValidator
cv = CrossValidator(estimator=dt, estimatorParamMaps=param_grid, evaluator=evaluator_auroc, numFolds=4)

with mlflow.start_run(experiment_id = experiment.experiment_id, run_name="best_model"):
    cv_model = cv.fit(training)
  
    mlflow.log_param("depth", cv_model.bestModel.depth)
    
    res = cv_model.bestModel.transform(test)

    test_metric_auroc = evaluator_auroc.evaluate(res)
    test_metric_acc = evaluator_acc.evaluate(res)
    test_metric_recall = evaluator_recall.evaluate(res)
    test_metric_prec = evaluator_prec.evaluate(res)
    test_metric_f = evaluator_f.evaluate(res)

    mlflow.log_metric(evaluator_auroc.getMetricName(), test_metric_auroc) 
    mlflow.log_metric(evaluator_acc.getMetricName(), test_metric_acc) 
    mlflow.log_metric(evaluator_recall.getMetricName(), test_metric_recall) 
    mlflow.log_metric(evaluator_prec.getMetricName(), test_metric_prec)     
    mlflow.log_metric(evaluator_f.getMetricName(), test_metric_f) 
  
    mlflow.spark.log_model(spark_model=cv_model.bestModel, artifact_path='best_classifier')<jupyter_output><empty_output><jupyter_text>ML Flow: logowanie eksperymentu z modelem GBT<jupyter_code>from pyspark.ml.classification import GBTClassifier
gbt = GBTClassifier(labelCol="label", featuresCol="features", maxIter=10)
gbt_model = gbt.fit(training)

with mlflow.start_run(experiment_id = experiment.experiment_id, run_name="gbt_model"):
  
    mlflow.log_param("depth", gbt.getMaxDepth())

    res = gbt_model.transform(test)
    
    test_metric_auroc = evaluator_auroc.evaluate(res)
    test_metric_acc = evaluator_acc.evaluate(res)
    test_metric_recall = evaluator_recall.evaluate(res)
    test_metric_prec = evaluator_prec.evaluate(res)
    test_metric_f = evaluator_f.evaluate(res)

    mlflow.log_metric(evaluator_auroc.getMetricName(), test_metric_auroc) 
    mlflow.log_metric(evaluator_acc.getMetricName(), test_metric_acc) 
    mlflow.log_metric(evaluator_recall.getMetricName(), test_metric_recall) 
    mlflow.log_metric(evaluator_prec.getMetricName(), test_metric_prec)     
    mlflow.log_metric(evaluator_f.getMetricName(), test_metric_f) 
  
    mlflow.spark.log_model(spark_model=gbt_model, artifact_path='gbt_classifier')<jupyter_output><empty_output><jupyter_text>ML Flow: serwowanie modeluZeby zapewnić prosty interfejs do klasyfikatora, zapiszemy model wraz z krokami wstępnego przetwarzania (przekształcenie danych wejściowych w wektor cech)<jupyter_code>import mlflow
import mlflow.spark
from pyspark.ml.classification import GBTClassifier

raw_training, raw_test = spark_df.randomSplit([0.8, 0.2], seed=1234)

full_classifier_name="full_gbt_classifier"
version=1

with mlflow.start_run(experiment_id = experiment.experiment_id, run_name="gbt_model_raw"):
    pipeline = Pipeline(stages=stringindexer_stages + \
                          onehotencoder_stages + \
                          [vectorassembler_stage] + [gbt] )
    model = pipeline.fit(raw_training)
    mlflow.spark.log_model(spark_model=model, artifact_path='gbt_classifier', registered_model_name=full_classifier_name)<jupyter_output><empty_output><jupyter_text>Klasyfikacja jako funkcja<jupyter_code>import mlflow.pyfunc
from pyspark.sql.functions import struct

registry_uri=f"models:/{full_classifier_name}/{version}"
pyfunc_udf = mlflow.pyfunc.spark_udf(spark, model_uri=registry_uri)<jupyter_output><empty_output><jupyter_text>Przykładowe wywołanie funkcji<jupyter_code>predicted_df = raw_test.limit(10)\
    .withColumn("prediction", pyfunc_udf(struct('OpSys', 'EdLevel', 'MainBranch' , 'Country', 'JobSeek', 'YearsCode'))) \
    .select ('OpSys', 'EdLevel', 'MainBranch' , 'Country', 'JobSeek', 'YearsCode', 'prediction')
predicted_df.toPandas()<jupyter_output><empty_output><jupyter_text>Klasyfikacja jako serwis RESTSekwencja komend do uruchomienia w terminalu```bashunset PYSPARK_SUBMIT_ARGSsource /opt/conda/etc/profile.d/conda.shconda activate $HOME/venv/$JUPYTER_KERNEL_NAMEmlflow models serve -m models:/full_gbt_classifier/1 -p 9090  --no-conda```  Przykładowe wywołanie serwisu<jupyter_code>import requests 

url = "http://localhost:9090/invocations"
headers = {'Content-Type': 'application/json; format=pandas-split'}

input_data = '{"columns":["OpSys", "EdLevel", "MainBranch" , "Country", "JobSeek", "YearsCode"], \
    "data":[[ \
    "MacOS",\
    "Master’s degree (M.A., M.S., M.Eng., MBA, etc.)",\
    "I am a developer by profession","United Kingdom",\
    "I am not interested in new job opportunities",\
    "10"]]}'\
    .encode('utf-8')

r = requests.post(url, data=input_data, headers=headers) 
r.text<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.10790806388780678, 'fraction_numerical': 0.0053564472146474485, 'mean_word_length': 6.573008849557522, 'pattern_counts': {'":': 2, '<': 45, '<?xml version=': 0, '>': 46, 'https://': 1, 'lorem ipsum': 0, 'www.': 1, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '11', 'dir': 'jupyter-structured-clean-dedup', 'id': '617637', 'license': 'Apache-2.0', 'n_tokens_mistral': 3722, 'n_tokens_neox': 3661, 'n_words': 577, 'path': 'session_3/mlflow.ipynb', 'repo_name': 'biodatageeks/ds-notebooks', 'size': '10268'}