We can create gcp agent platform pipeline using kubeflow or tensorflow SDK (this is for large dataset which span terrabytes). Kubeflow sdk are basically python script that you defines pipelines or components. @kfp.dsl.pipeline and @kfp.dsl.componet.
Lets gets started by installing the required modules
uv install kfp.dsl google-cloud-pipeline-components docker
In its simplistic form, a pipeline can look like this (this will not work for Windows setup with docker)
local_pipeline.py
from kfp import dsl
# Define a component to add two numbers.
@dsl.component
def add(a: int, b: int) -> int:
return a + b
# Define a simple pipeline using the component.
@dsl.pipeline
def addition_pipeline(x: int, y: int, z: int) -> int:
task1 = add(a=x, b=y)
task2 = add(a=task1.output, b=z)
return task2.output
And here is our runner
from kfp import local
from local_pipeline import addition_pipeline
local.init(runner=local.DockerRunner())
pipeline_task = addition_pipeline(x=1, y=2, z=3)
print(f'Result: {pipeline_task.output}')
And then running the pipeline locally - please ensure you started your docker instance.
And a more realistic workflow would look like this. At the bottom of the code, you will notice that I have compile which turn the pipeline into a yaml code that we can deploy to Google Agent Platform.
Important thing to note here is always use google_cloud_pipeline_components to build a pipeline.
import kfp
from google.cloud import aiplatform
from google_cloud_pipeline_components.v1.dataset import ImageDatasetCreateOp
from google_cloud_pipeline_components.v1.automl.training_job import AutoMLImageTrainingJobRunOp
from google_cloud_pipeline_components.v1.endpoint import EndpointCreateOp, ModelDeployOp
project_id = "project-id"
pipeline_root_path = "./"
# Define the workflow of the pipeline.
@kfp.dsl.pipeline(
name="automl-image-training-v2",
pipeline_root=pipeline_root_path)
def pipeline(project_id: str):
# The first step of your workflow is a dataset generator.
# This step takes a Google Cloud Pipeline Component, providing the necessary
# input arguments, and uses the Python variable `ds_op` to define its
# output. Note that here the `ds_op` only stores the definition of the
# output but not the actual returned object from the execution. The value
# of the object is not accessible at the dsl.pipeline level, and can only be
# retrieved by providing it as the input to a downstream component.
ds_op = ImageDatasetCreateOp(
project=project_id,
display_name="flowers",
gcs_source="gs://cloud-samples-data/vision/automl_classification/flowers/all_data_v2.csv",
import_schema_uri=aiplatform.schema.dataset.ioformat.image.single_label_classification,
)
# The second step is a model training component. It takes the dataset
# outputted from the first step, supplies it as an input argument to the
# component (see `dataset=ds_op.outputs["dataset"]`), and will put its
# outputs into `training_job_run_op`.
training_job_run_op = AutoMLImageTrainingJobRunOp(
project=project_id,
display_name="train-iris-automl-mbsdk-1",
prediction_type="classification",
model_type="CLOUD",
dataset=ds_op.outputs["dataset"],
model_display_name="iris-classification-model-mbsdk",
training_fraction_split=0.6,
validation_fraction_split=0.2,
test_fraction_split=0.2,
budget_milli_node_hours=8000,
)
# The third and fourth step are for deploying the model.
create_endpoint_op = EndpointCreateOp(
project=project_id,
display_name = "create-endpoint",
)
model_deploy_op = ModelDeployOp(
model=training_job_run_op.outputs["model"],
endpoint=create_endpoint_op.outputs['endpoint'],
automatic_resources_min_replica_count=1,
automatic_resources_max_replica_count=1,
)
from kfp import compiler
compiler.Compiler().compile(
pipeline_func=pipeline,
package_path='image_classif_pipeline.yaml'
)
ImageDatasetCreateOp - This component creates a managed Vertex AI Image Dataset and imports your data into the platform. It handles the cloud plumbing required to read files out of Google Cloud Storage and index them for training.
AutoMLImageTrainingJobRunOp - This component kicks off a serverless Google AutoML training job. It consumes the managed dataset generated by the previous step, then handles the heavy lifting of architecture search, data splitting, and hyperparameter tuning entirely in the background
EndpointCreateOp - It creates an endpoint and with this, we can deploy multiple version of our model and ensure production model are not impacted when we are 'trialing' new models. We can configure traffic splitting between the models.
ModelDeployOps - is the final, crucial component from the google_cloud_pipeline_components library that takes your trained model and pushes it live onto a provisioned hosting environment.
While EndpointCreateOp sets up the empty network router and URL slot, ModelDeployOp does the heavy lifting of pulling your model artifacts, provisioning the physical virtual machines or GPUs, spinning up the model server containers, and mapping them to that endpoint.
And this is a sample of what the pipeline would look like :-
Comments