Google Cloud Pipeline コンポーネント（GCPC）を使用すると、次の Vertex AI と Google Cloud の機能を使用してコンポーネントとアーティファクトを保護できます。
コンポーネントのサービス アカウントを指定する
コンポーネントを使用するときに、サービス アカウントを指定することもできます。コンポーネントが起動し、このサービス アカウントの権限で動作します。たとえば、次のコードを使用して、ModelDeploy コンポーネントのサービス アカウントを指定できます。
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")
次のように置き換えます。
- SERVICE_ACCOUNT_ID: サービス アカウントの ID。
- PROJECT_ID: プロジェクトの ID。
カスタム サービス アカウントの使用と Vertex AI Pipelines で使用するサービス アカウントの構成の詳細を確認してください。
VPC Service Controls を使用してデータ漏洩を防ぐ
VPC Service Controls を使用することで、Vertex AI Pipelines からのデータ漏洩のリスクを軽減できます。VPC Service Controls を使用してサービス境界を作成すると、Vertex AI Pipelines と Google Cloud Pipeline コンポーネントによって作成されるリソースとデータが自動的に保護されます。たとえば、VPC Service Controls を使用してパイプラインを保護する場合、次のアーティファクトがサービス境界を離れることはできません。
- AutoML モデルのトレーニング データ
- 作成したモデル
- バッチ予測リクエストの結果
Vertex AI を使用した VPC Service Controls の詳細を確認してください。
VPC ネットワーク ピアリングを設定する
追加のパラメータを指定することで、Virtual Private Cloud とピアリングするように Google Cloud Pipeline コンポーネントを構成できます。たとえば、次のコードを使用して、EndpointCreate コンポーネントの VPC ネットワークを指定します。
endpoint_create_op = EndpointCreateOp(
project="PROJECT_ID",
location="REGION",
display_name="endpoint-display-name",
network="NETWORK")
次のように置き換えます。
- PROJECT_ID: プロジェクトの ID。
- REGION: Vertex AI を使用するリージョン。
- NETWORK: VPC ネットワーク（例:
"projects/12345/global/networks/myVPC"）。
Vertex AI での VPC ネットワーク ピアリングの詳細を確認してください。
顧客管理の暗号鍵（CMEK）を使用する
Google Cloud のデフォルトでは、Google が管理する暗号鍵を使用して、保存されているデータを自動的に暗号化します。データを保護する鍵について特定のコンプライアンスや規制の要件がある場合は、リソースで顧客管理の暗号鍵（CMEK）を使用できます。顧客管理の暗号鍵の使用を開始する前に、Vertex AI での CMEK のメリットと現在 CMEK でサポートされているリソースを確認してください。
CMEK を使用したコンポーネントの構成
Cloud Key Management Service でキーリングと鍵を作成し、鍵に対する Vertex AI の暗号化と復号の権限を付与したら、create パラメータの一つとしてキーを指定して CMEK でサポートされる新しいコンポーネントを作成できます。たとえば、次のコードを使用して ModelBatchPredict コンポーネントのキーを指定します。
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")
次のように置き換えます。
- PROJECT_ID: Google Cloud プロジェクト ID。
- LOCATION_ID: 有効なロケーションまたはリージョン ID（例:
us-central1）。 - KEY_RING_NAME: CMEK のキーリングの名前。キーリングの詳細については、Cloud KMS リソースをご覧ください。
- KEY_NAME: CMEK 鍵名。
注: Vertex AI コンポーネント以外の Google Cloud コンポーネントでは、追加の権限が必要になる場合があります。たとえば、BigQuery コンポーネントでは暗号化と復号の権限が必要になる場合があります。また、CMEK 鍵の場所は、コンポーネントと同じ場所にする必要があります。たとえば、BigQuery コンポーネントがマルチリージョンの米国ロケーションにあるデータセットからデータを読み込む場合は、CMEK 鍵もマルチリージョンの米国ロケーションに基づいている必要があります。
コンポーネントでアーティファクトを使用または生成する
Google Cloud Pipeline コンポーネント SDK は、コンポーネントの入力および出力として機能する一連の ML メタデータ アーティファクト タイプを定義します。一部の Google Cloud パイプライン コンポーネントは、これらのアーティファクトを入力として使用するか、出力として生成します。
このページでは、これらのアーティファクトの使用と生成の方法について説明します。
ML アーティファクトを使用する
コンポーネント YAML でアーティファクトを使用する
アーティファクトのメタデータは、コンポーネントへの入力として機能します。アーティファクトを入力として使用するには、それを抽出してコンポーネントの YAML ファイルに含める必要があります。
たとえば、ModelUploadOp コンポーネントによって生成される google.VertexModel アーティファクトは、ModelDeployOp コンポーネントで使用できます。コンポーネントの YAML ファイルで次のコードを使用し、入力（参照）から Vertex AI の Model リソースを取得します。
"model": "',"{{$.inputs.artifacts['model'].metadata['resourceName']}}", '"'アーティファクト メタデータの完全なスキーマについては、Kubeflow GitHub リポジトリの artifact_types.py ファイルをご覧ください。
軽量の Python コンポーネントでアーティファクトを使用する
from kfp.dsl import Artifact, Input
@dsl.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"]
Vertex ML Metadata アーティファクト タイプの使用方法については、表形式データと Vertex AI AutoML を使用して分類モデルをトレーニングするをご覧ください。
ML アーティファクトを作成する
次のコードサンプルは、GCPC コンポーネント入力として受け入れ可能な Vertex ML Metadata アーティファクトの作成方法を示しています。
インポータ ノードを使用する
次の例では、Vertex ML Metadata に新しいアーティファクト エントリを登録するインポータ ノードを作成します。インポータ ノードは、アーティファクトの URI とメタデータをプリミティブとして受け入れ、アーティファクトにパッケージ化します。
from google_cloud_pipeline_components import v1
from google_cloud_pipeline_components.types import artifact_types
from kfp.components import importer_node
from kfp 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 = v1.model.ModelUploadOp(
project=_GCP_PROJECT_ID,
location=_GCP_REGION,
display_name=_MODEL_DISPLAY_NAME,
unmanaged_container_model=importer_spec.outputs['artifact'])
Python 関数ベースのコンポーネントを使用する
次の例は、Python コンポーネントから Vertex ML Metadata アーティファクトを直接出力する方法を示しています。
from google_cloud_pipeline_components import v1
from kfp.components import importer_node
from kfp 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 = v1.model.ModelUploadOp(
project=_GCP_PROJECT_ID,
location=_GCP_REGION,
display_name=_MODEL_DISPLAY_NAME,
unmanaged_container_model=unmanaged_model_op.outputs['model'])
独自のコンテナベースのコンポーネントを使用する
次の例は、artifact_types.py ユーティリティ クラスを使用してコンテナベースのコンポーネントからの出力として VertexBatchPredictionJob アーティファクトを生成する方法を示しています。
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])