When you use Google Cloud Pipeline Components (GCPC), you can use the following Vertex AIand 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 ofaModelDeploy 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 aboutusing a custom service accountandconfiguring a service accountfor use with Vertex AI Pipelines.

Use VPC Service Controls to prevent data exfiltration

VPC Service Controls can help you mitigatethe risk of data exfiltration from Vertex AI Pipelines. When you useVPC Service Controls to create a service perimeter, resources and data that arecreated by Vertex AI Pipelines and the Google Cloud Pipeline Components are automaticallyprotected. For example, when you use VPC Service Controls to protect yourpipeline, the following artifacts can't leave your service perimeter:

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

Learn more aboutVPC 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 extraparameters. For example, you can use the following code to specify aVPC network for anEndpointCreate 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 aboutVPC Network Peering in Vertex AI.

Use customer-managed encryption keys (CMEK)

By default, Google Cloud automaticallyencrypts data when atrest using encryption keysmanaged by Google. If you have specific compliance or regulatory requirementsrelated to the keys that protect your data, you can use customer-managedencryption keys (CMEK) for your resources. Before you start to usecustomer-managed encryption keys, learn about thebenefits of CMEK on Vertex AIandcurrent CMEK supported resources.

Configuring your component with CMEK

After you create a key ring and key inCloud Key Management Service,and grant Vertex AI encrypter and decrypter permissions for your key,you can create a new CMEK-supported component by specifying your key as one ofthe create parameters. For example, you can use the following code to specifya key for aModelBatchPredict 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 Google Cloud 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, seeCloud KMS resources.
• KEY_NAME: The CMEK key name.

Note: Google Cloud components that aren't Vertex AI componentsmight require additional permissions. For example, a BigQuerycomponent might requireencryption and decryption permission.In addition, the location of the CMEK key must be the same as the location ofthe component. For example, if a BigQuery component loads data froma dataset that's based in themulti-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 SDK defines a set ofML metadata artifact types that serve as componentinput and output. Some Google Cloud Pipeline Components consume these artifacts as inputor 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 anartifact to be consumed as input, you must extract it and put it in a componentYAML file.

For example, theModelUploadOp component generates agoogle.VertexModel artifact which can be consumed by aModelDeployOp component. Use the following code in a component YAML file to retrieve thea Vertex AIModel resource from the inputs (reference):

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

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

Consume an artifact in a lightweight Python component

fromkfp.dslimportArtifact,Input@dsl.componentdefclassification_model_eval_metrics(project:str,location:str,# "us-central1",model:Input[Artifact],):# Consumes the `resourceName` metadatamodel_resource_path=model.metadata["resourceName"]

For an example of how to consume the Vertex ML Metadata artifactstypes,seeTrain a classification model using tabular data and Vertex AI AutoML.

Create an ML artifact

The following code examples show how to create a Vertex ML Metadataartifact that a GCPC component can accept as input.

Use an importer node

The following example creates an Importer node that registers a new artifactentry to Vertex ML Metadata. The importer node takes the artifact'sURI and metadataas primitives and packages them into an artifact.

fromgoogle_cloud_pipeline_componentsimportv1fromgoogle_cloud_pipeline_components.typesimportartifact_typesfromkfp.componentsimportimporter_nodefromkfpimportdsl@dsl.pipeline(name=_PIPELINE_NAME)defpipeline():# 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 stepmodel_upload_with_artifact_op=v1.model.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 Vertex ML Metadataartifactdirectly from a Python component.

fromgoogle_cloud_pipeline_componentsimportv1fromkfp.componentsimportimporter_nodefromkfpimportdsl@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.defreturn_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)defpipeline():unmanaged_model_op=return_unmanaged_model()# Consuming the UnmanagedContainerModel artifact for the previous stepmodel_upload_with_artifact_op=v1.model.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 aVertexBatchPredictionJob artifact as output from acontainer-based component using theartifact_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])