, "y_train"],

Annotated[pd.Series, "y_test"],

]:"""Load the iris dataset as tuple of Pandas DataFrame / Series."""

iris = load_iris(as_frame=True)

X_train, X_test, y_train, y_test = train_test_split(

iris.data, iris.target, test_size=0.2, shuffle=True, random_state=42

return X_train, X_test, y_train, y_test

@step

def svc_trainer(

X_train: pd.DataFrame,

y_train: pd.Series,

gamma: float = 0.001,

) -> Tuple[

Annotated[ClassifierMixin, "trained_model"],

Annotated[float, "training_acc"],

]:

"""Train a sklearn SVC classifier and log to MLflow."""

model = SVC(gamma=gamma)

model.fit(X_train.to_numpy(), y_train.to_numpy())

train_acc = model.score(X_train.to_numpy(), y_train.to_numpy())

print(f"Train accuracy: {train_acc}")

return model, train_acc

@pipeline

def training_pipeline(gamma: float = 0.002):

X_train, X_test, y_train, y_test = training_data_loader()

svc_trainer(gamma=gamma, X_train=X_train, y_train=y_train)

if __name__ == "__main__":

training_pipeline()

# Step one will use cache, step two will rerun.

# ZenML will detect a different value for the

# `gamma` input of the second step and disable caching.

logger.info("\n\nFirst step cached, second not due to parameter change")

training_pipeline(gamma=0.0001)

# This will disable cache for the second step.

logger.info("\n\nFirst step cached, second not due to settings")

svc_trainer = svc_trainer.with_options(enable_cache=False)

training_pipeline()

# This will disable cache for all steps.

logger.info("\n\nCaching disabled for the entire pipeline")

training_pipeline.with_options(enable_cache=False)()

PreviousCreate an ML pipeline

NextManage artifacts

Last updated 8 days ago