Use Google Cloud Pipeline Components

When you use Google Cloud Pipeline Components, you can use the following Vertex AI and Google Cloud features to secure your components and artifacts.

Specify a service account for a component

When you use a component, you can optionally specify a service account. Your component launches and acts with the permissions of this service account. For example, you can use the following code to specify the service account of a ModelDeploy component:

model_deploy_op = ModelDeployOp(model=training_job_run_op.outputs["model"],
    endpoint=endpoint_op.outputs["endpoint"],
    automatic_resources_min_replica_count=1,
    automatic_resources_max_replica_count=1,
    service_account="SERVICE_ACCOUNT_ID@PROJECT_ID.iam.gserviceaccount.com")

Replace the following:

  • SERVICE_ACCOUNT_ID: The ID for the service account.
  • PROJECT_ID: The ID of the project.

Learn more about using a custom service account and configuring a service account for use with Vertex AI Pipelines.

Use VPC Service Controls to prevent data exfiltration

VPC Service Controls can help you mitigate the risk of data exfiltration from Vertex AI Pipelines. When you use VPC Service Controls to create a service perimeter, resources and data that are created by Vertex AI Pipelines and the Google Cloud Pipeline Components are automatically protected. For example, when you use VPC Service Controls to protect your pipeline, the following artifacts cannot leave your service perimeter:

  • Training data for an AutoML model
  • Models that you created
  • Results from a batch prediction request

Learn more about VPC Service Controls with Vertex AI.

Set up VPC Network Peering

You can configure Google Cloud Pipeline Components to peer with a Virtual Private Cloud by providing extra parameters. For example, you can use the following code to specify a VPC network for an EndpointCreate component:

endpoint_create_op = EndpointCreateOp(
    project="PROJECT_ID",
    location="REGION",
    display_name="endpoint-display-name",
    network="NETWORK")

Replace the following:

  • PROJECT_ID: The ID of the project.
  • REGION: The region where you are using Vertex AI.
  • NETWORK: The VPC network, for example, "projects/12345/global/networks/myVPC".

Learn more about VPC network peering in Vertex AI.

Use customer-managed encryption keys (CMEK)

By default, Google Cloud automatically encrypts data when it is at rest using encryption keys managed by Google. If you have specific compliance or regulatory requirements related to the keys that protect your data, you can use customer-managed encryption keys (CMEK) for your resources. Before you start to use customer-managed encryption keys, learn about the benefits of CMEK on Vertex AI and current CMEK supported resources.

Configuring your component with CMEK

After you create a keyring and key in Cloud Key Management Service, grant Vertex AI encrypter and decrypter permissions for your key, you can create a new CMEK-supported component by specifying your key as one of the create parameters. For example, you can use the following code to specify a key for a ModelBatchPredict component:

model_batch_predict_op = ModelBatchPredictOp(project="PROJECT_ID",
    model=model_upload_op.outputs["model"],
    encryption_spec_key_name="projects/PROJECT_ID/locations/LOCATION_ID/keyRings/KEY_RING_NAME/cryptoKeys/KEY_NAME")

Replace the following:

  • PROJECT_ID: Your GCP project ID.
  • LOCATION_ID: A valid location or region identifier, for example, us-central1.
  • KEY_RING_NAME: The name of the key ring for your CMEK. For more information about key rings, see Cloud KMS resources.
  • KEY_NAME: The CMEK key name.

Note: Google Cloud components that aren't Vertex AI components might require additional permissions. For example, a BigQuery component might require encryption and decryption permission. In addition, the location of the CMEK key must be the same as the location of the component. For example, if a BigQuery component loads data from a dataset that's based in the multi-region US location, the CMEK key must also be based in the multi-region US location.

Consume or produce artifacts in your component

The Google Cloud Pipeline Components SDK defines a set of ML metadata artifact types that serve as component input and output. Some Google Cloud Pipeline Components consume these artifacts as input or produce them as output.

This page shows how to consume and produce these artifacts.

Consume an ML artifact

Consume an artifact in component YAML

The artifact's metadata can serve as input to a component. To prepare an artifact to be consumed as input, you must extract it and put it in a component YAML file.

For instance, the ModelUploadOp component generates a google.VertexModel artifact which can be consumed by a ModelDeployOp component. Use the following code in a component YAML file to retrieve the a Vertex AI Model resource from the inputs (reference):

"model": "',"{{$.inputs.artifacts['model'].metadata['resourceName']}}", '"'

For the complete schema of the artifact's metadata, see the artifact_types.py file in the Kubeflow GitHub repo.

Consume an artifact in a lightweight Python component

from kfp.v2.dsl import (Artifact, component)
@component
def classification_model_eval_metrics(
    project: str,
    location: str,  # "us-central1",
    model: Input[Artifact],
) :
   # Consumes the `resourceName` metadata
   model_resource_path = model.metadata["resourceName"]

For an example of how to consume the Google ML Metadata artifacts types, see Train a classification model using tabular data and Vertex AI AutoML.

Create an ML artifact

The following code examples show how to create a Google ML Metadata artifact that a GCPC component can accept as input.

Use an importer node

The following example creates an Importer node that registers a new artifact entry to ML Metadata. The importer node takes the artifact's URI and metadata as primitives and packages them into an artifact.

from google_cloud_pipeline_components import aiplatform as gcpc_aip
from google_cloud_pipeline_components.types import artifact_types
from kfp.v2.components import importer_node
from kfp.v2 import dsl

@dsl.pipeline(name=_PIPELINE_NAME)
def pipeline():
  # Using importer and UnmanagedContainerModel artifact for model upload
  # component.
  importer_spec = importer_node.importer(
      artifact_uri='gs://managed-pipeline-gcpc-e2e-test/automl-tabular/model',
      artifact_class=artifact_types.UnmanagedContainerModel,
      metadata={
          'containerSpec': {
              'imageUri':
                  'us-docker.pkg.dev/vertex-ai/automl-tabular/prediction-server:prod'
          }
      })

  # Consuming the UnmanagedContainerModel artifact for the previous step
  model_upload_with_artifact_op = gcpc_aip.ModelUploadOp(
      project=_GCP_PROJECT_ID,
      location=_GCP_REGION,
      display_name=_MODEL_DISPLAY_NAME,
      unmanaged_container_model=importer_spec.outputs['artifact'])

Use Python function-based components

The following example shows how to output a Google ML Metadata artifact directly from a Python component.

from google_cloud_pipeline_components import aiplatform as gcpc_aip
from google_cloud_pipeline_components.types import artifact_types
from kfp.v2.components import importer_node
from kfp.v2 import dsl

@dsl.component(
    base_image='python:3.9',
    packages_to_install=['google-cloud-aiplatform'],
)
# Note currently KFP SDK doesn't support outputting artifacts in `google` namespace.
# Use the base type dsl.Artifact instead.
def return_unmanaged_model(model: dsl.Output[dsl.Artifact]):
  model.metadata['containerSpec'] = {
      'imageUri':
          'us-docker.pkg.dev/vertex-ai/automl-tabular/prediction-server:prod'
  }
  model.uri = f'gs://automl-tabular-pipeline/automl-tabular/model'

@dsl.pipeline(name=_PIPELINE_NAME)
def pipeline():

  unmanaged_model_op = return_unmanaged_model()

  # Consuming the UnmanagedContainerModel artifact for the previous step
  model_upload_with_artifact_op = gcpc_aip.ModelUploadOp(
      project=_GCP_PROJECT_ID,
      location=_GCP_REGION,
      display_name=_MODEL_DISPLAY_NAME,
      unmanaged_container_model=unmanaged_model_op.outputs['model'])

Use your own container-based component

The following example shows how to generate an artifact as output from a container-based component using the artifact_types.py utility class.

bp_job_artifact = VertexBatchPredictionJob(
    'batchpredictionjob', vertex_uri_prefix + get_job_response.name,
    get_job_response.name, get_job_response.output_info.bigquery_output_table,
    get_job_response.output_info.bigquery_output_dataset,
    get_job_response.output_info.gcs_output_directory)

    output_artifacts = executor_input_json.get('outputs', {}).get('artifacts', {})
    executor_output['artifacts'] = bp_job_artifact.to_executor_output_artifact(output_artifacts[bp_job_artifact.name])

For a concrete example, see Vertex Batch Prediction Job component.