Pipeline

Machine Learning with PySpark

Andrew Collier

Data Scientist, Fathom Data

Data Leakage?

เมธอด fit()

ใช้กับข้อมูล training เท่านั้น

เมธอด transform()

ใช้กับข้อมูล testing และ training

โมเดลที่มี Data Leakage

โมเดลที่ใช้ข้อมูล testing ในการ training

โมเดลที่ไม่มี Data Leakage

โมเดลที่ใช้เฉพาะข้อมูล training ในการ training

Pipeline

Pipeline คือชุดของขั้นตอนที่ทำงานต่อเนื่องกัน

Pipeline ที่มีหลายขั้นตอน

จะรันทีละขั้นตอนก็ได้ หรือจะใช้ pipeline รันทีเดียวก็ได้!

โมเดลรถยนต์: ขั้นตอน

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

โมเดลรถยนต์: การใช้งานแต่ละขั้นตอน

ข้อมูล Training

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)

ข้อมูล Testing

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)

โมเดลรถยนต์: Pipeline

รวมทุกขั้นตอนเข้าเป็น pipeline เดียว

from pyspark.ml import Pipeline

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

ข้อมูล Training

pipeline = pipeline.fit(cars_train)

ข้อมูล Testing

predictions = pipeline.transform(cars_test)

โมเดลรถยนต์: Stages

เข้าถึงแต่ละขั้นตอนได้ผ่าน attribute .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])

Pipeline ช่วยให้ workflow คล่องตัวขึ้น!

Machine Learning with PySpark