Pipeline

Maskininlärning med PySpark

Andrew Collier

Data Scientist, Fathom Data

Dataläckage?

Metoden fit()

Enbart för träningsdata.

Metoden transform()

För test- och träningsdata.

En modell med dataläckage

Modell där testdata används vid träning

En tät modell

Modell där enbart träningsdata används vid träning

Pipeline

En pipeline består av en serie operationer.

En pipeline med flera steg

Du kan tillämpa varje operation separat... eller helt enkelt köra hela pipelinen!

Bilmodell: Steg

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

Bilmodell: Tillämpa stegen

Träningsdata

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)

Testdata

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)

Bilmodell: Pipeline

Kombinera stegen i en pipeline.

from pyspark.ml import Pipeline

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

Träningsdata

pipeline = pipeline.fit(cars_train)

Testdata

predictions = pipeline.transform(cars_test)

Bilmodell: Steg

Kom åt enskilda steg via attributet .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])

Pipelines effektiviserar arbetsflödet!

Maskininlärning med PySpark