Pipeline

Machine Learning avec PySpark

Andrew Collier

Data Scientist, Fathom Data

Fuite de données ?

La méthode fit()

Uniquement pour les données d'entraînement.

La méthode transform()

Pour les données d'essai et d'entraînement.

Machine Learning avec PySpark

Un modèle avec fuite

Modèle où les données d'essai servent à l'entraînement

Machine Learning avec PySpark

Un modèle étanche

Modèle où seules les données d'entraînement servent à l'entraînement

Machine Learning avec PySpark

Pipeline

Un pipeline regroupe une série d'opérations.

Un pipeline à plusieurs étapes

Vous pourriez appliquer chaque opération séparément… ou simplement exécuter le pipeline !

Machine Learning avec PySpark

Modèle « Cars » : étapes

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 avec PySpark

Modèle « Cars » : appliquer les étapes

Données d'entraînement

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)

# Ajuster le modèle sur les données d'entraînement
regression = regression.fit(cars_train)

Données d'essai

cars_test  = indexer.transform(cars_test)

cars_test  = onehot.transform(cars_test)

cars_test  = assemble.transform(cars_test)

# Générer des prédictions sur les données d'essai
predictions = regression.transform(cars_test)
Machine Learning avec PySpark

Modèle « Cars » : pipeline

Regrouper les étapes dans un pipeline.

from pyspark.ml import Pipeline

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

Données d'entraînement

pipeline = pipeline.fit(cars_train)

Données d'essai

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

Modèle « Cars » : étapes

Accéder aux étapes via l'attribut .stages.

# L'objet LinearRegression (quatrième étape -> 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 avec PySpark

Les pipelines simplifient le flux de travail !

Machine Learning avec PySpark

Preparing Video For Download...