Vertex AI Pipelines Advanced

from kfp import dsl
from kfp.v2 import compiler
from google.cloud import aiplatform

# Define pipeline components
@dsl.component(base_image="python:3.10", packages_to_install=["pandas", "scikit-learn"])
def preprocess_data(input_path: str, output_path: dsl.OutputPath("Dataset")):
    import pandas as pd
    df = pd.read_csv(input_path)
    # Preprocessing logic here
    df.to_csv(output_path, index=False)

@dsl.component(base_image="python:3.10", packages_to_install=["scikit-learn", "joblib"])
def train_model(dataset: dsl.InputPath("Dataset"), model_path: dsl.OutputPath("Model")):
    from sklearn.ensemble import RandomForestClassifier
    import joblib, pandas as pd
    df = pd.read_csv(dataset)
    X, y = df.drop("target", axis=1), df["target"]
    model = RandomForestClassifier().fit(X, y)
    joblib.dump(model, model_path)

# Define the pipeline
@dsl.pipeline(name="ml-training-pipeline", description="End-to-end ML pipeline")
def ml_pipeline(input_data: str):
    preprocess_task = preprocess_data(input_path=input_data)
    train_task = train_model(dataset=preprocess_task.outputs["output_path"])

# Compile and run
compiler.Compiler().compile(pipeline_func=ml_pipeline, package_path="pipeline.json")

aiplatform.init(project="my-project", location="us-central1")
job = aiplatform.PipelineJob(
    display_name="my-pipeline-run",
    template_path="pipeline.json",
    parameter_values={"input_data": "gs://my-bucket/data.csv"}
)
job.run()

from google.cloud import aiplatform

# Create a Feature Store
featurestore = aiplatform.Featurestore.create(
    featurestore_id="my_featurestore",
    online_store_fixed_node_count=1
)

# Create an entity type
entity_type = featurestore.create_entity_type(
    entity_type_id="users",
    description="User features"
)

# Add features
entity_type.create_feature(feature_id="age", value_type="INT64")
entity_type.create_feature(feature_id="purchase_count", value_type="INT64")
entity_type.create_feature(feature_id="avg_spend", value_type="DOUBLE")

# Ingest feature values from BigQuery
entity_type.ingest_from_bq(
    feature_ids=["age", "purchase_count", "avg_spend"],
    feature_time="timestamp",
    bq_source_uri="bq://my-project.dataset.features_table",
    entity_id_field="user_id"
)

# Upload a model to the registry
model = aiplatform.Model.upload(
    display_name="my-model",
    artifact_uri="gs://my-bucket/model-artifacts/",
    serving_container_image_uri="us-docker.pkg.dev/vertex-ai/prediction/sklearn-cpu.1-0:latest",
    labels={"team": "ml-engineering", "version": "1.0"},
    description="Random Forest classifier v1"
)

# List all model versions
models = aiplatform.Model.list(filter='display_name="my-model"')
for m in models:
    print(f"{m.display_name} - {m.resource_name}")

# Schedule a pipeline to run daily
from google.cloud.aiplatform import pipeline_jobs

schedule = aiplatform.PipelineJob.create_schedule(
    display_name="daily-retraining",
    cron="0 2 * * *",  # Run at 2 AM daily
    template_path="pipeline.json",
    parameter_values={"input_data": "gs://my-bucket/latest-data.csv"}
)