Pipeline

Machine Learning met PySpark

Andrew Collier

Data Scientist, Fathom Data

Leakage?

De methode fit()

Alleen voor trainingsdata.

De methode transform()

Voor test- en trainingsdata.

Een lek model

Model waarbij testdata voor training is gebruikt

Een waterdicht model

Model waarbij alleen trainingsdata is gebruikt

Pipeline

Een pipeline is een reeks bewerkingen.

Een pipeline met meerdere stages

Je kunt elke stap los toepassen... of gewoon de pipeline gebruiken!

Automodel: Stappen

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

Autonmodel: Stappen toepassen

Trainingsdata

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)

# Model fitten op trainingsdata
regression = regression.fit(cars_train)

Testdata

cars_test  = indexer.transform(cars_test)

cars_test  = onehot.transform(cars_test)

cars_test  = assemble.transform(cars_test)

# Voorspellingen maken op testdata
predictions = regression.transform(cars_test)

Automodel: Pipeline

Combineer stappen in een pipeline.

from pyspark.ml import Pipeline

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

Trainingsdata

pipeline = pipeline.fit(cars_train)

Testdata

predictions = pipeline.transform(cars_test)

Automodel: Stages

Toegang tot losse stages via de .stages-attribuut.

# Het LinearRegression-object (vierde 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])

Pipelines versnellen je workflow!

Machine Learning met PySpark