Pipeline

Machine Learning cu PySpark

Andrew Collier

Data Scientist, Fathom Data

Scurgere de date?

Metoda fit()

Doar pentru datele de antrenament.

Metoda transform()

Pentru datele de testare și de antrenament.

Machine Learning cu PySpark

Un model cu scurgere de date

Model în care datele de testare sunt folosite la antrenament

Machine Learning cu PySpark

Un model fără scurgere de date

Model în care doar datele de antrenament sunt folosite la antrenament

Machine Learning cu PySpark

Pipeline

Un pipeline constă dintr-o serie de operații.

Un pipeline cu mai multe etape

Puteți aplica fiecare operație individual... sau puteți aplica direct pipeline-ul!

Machine Learning cu PySpark

Modelul pentru mașini: Pași

indexer = StringIndexer(inputCol='type', outputCol='type_idx')

onehot = OneHotEncoder(inputCols=['type_idx'], outputCols=['type_dummy'])

assemble = VectorAssembler(
  inputCols=['mass', 'cyl', 'type_dummy'],
  outputCol='features'
)

regression = LinearRegression(labelCol='consumption')
Machine Learning cu PySpark

Modelul pentru mașini: Aplicarea pașilor

Date de antrenament

indexer = indexer.fit(cars_train)
cars_train = indexer.transform(cars_train)

onehot = onehot.fit(cars_train)
cars_train = onehot.transform(cars_train)

cars_train = assemble.transform(cars_train)

# Fit model to training data
regression = regression.fit(cars_train)

Date de testare

cars_test  = indexer.transform(cars_test)

cars_test  = onehot.transform(cars_test)

cars_test  = assemble.transform(cars_test)

# Make predictions on testing data
predictions = regression.transform(cars_test)
Machine Learning cu PySpark

Modelul pentru mașini: Pipeline

Combinați pașii într-un pipeline.

from pyspark.ml import Pipeline

pipeline = Pipeline(stages=[indexer, onehot, assemble, regression])

Date de antrenament

pipeline = pipeline.fit(cars_train)

Date de testare

predictions = pipeline.transform(cars_test)
Machine Learning cu PySpark

Modelul pentru mașini: Etape

Accesați etapele individuale folosind atributul .stages.

# The LinearRegression object (fourth stage -> index 3)
pipeline.stages[3]


print(pipeline.stages[3].intercept)

4.19433571782916

print(pipeline.stages[3].coefficients)

DenseVector([0.0028, 0.2705, -1.1813, -1.3696, -1.1751, -1.1553, -1.8894])
Machine Learning cu PySpark

Pipeline-urile simplifică fluxul de lucru!

Machine Learning cu PySpark

Preparing Video For Download...