Google Cloud AI Platform 运算符¶
Google Cloud AI Platform（以前称为 ML Engine）可用于大规模训练机器学习模型、在云端托管训练的模型以及使用模型对新数据进行预测。AI Platform 是一组用于训练、评估和调整机器学习模型的工具。AI Platform 还可用于部署训练的模型、进行预测和管理各种模型版本。
AI Platform Training、AI Platform Prediction、AI Platform Pipelines 和 AI Platform Data Labeling Service 的旧版本已弃用，在关闭日期后将不再在 Google Cloud 上提供。旧版 AI Platform 的所有功能和新功能均可在 Vertex AI 平台上使用。
先决任务¶
要使用这些运算符，您必须执行以下操作
使用 Cloud 控制台 选择或创建 Cloud Platform 项目。
为您的项目启用帐单，如 Google Cloud 文档 中所述。
启用 API，如 Cloud 控制台文档 中所述。
通过pip安装 API 库。
pip install 'apache-airflow[google]'有关详细信息，请参阅 安装。
启动作业¶
要使用 AI Platform 启动机器学习操作，您必须启动训练作业。这会创建一个虚拟机，该虚拟机可以运行训练器文件中指定且包含主应用程序代码的代码。可以使用 MLEngineStartTrainingJobOperator 启动作业。
此操作已弃用。请改用 CreateCustomPythonPackageTrainingJobOperator。
create_custom_python_package_training_job = CreateCustomPythonPackageTrainingJobOperator(
task_id="create_custom_python_package_training_job",
staging_bucket=f"gs://{CUSTOM_PYTHON_GCS_BUCKET_NAME}",
display_name=PACKAGE_DISPLAY_NAME,
python_package_gcs_uri=PYTHON_PACKAGE_GCS_URI,
python_module_name=PYTHON_MODULE_NAME,
container_uri=TRAIN_IMAGE,
model_serving_container_image_uri=DEPLOY_IMAGE,
bigquery_destination=f"bq://{PROJECT_ID}",
# run params
dataset_id=tabular_dataset_id,
model_display_name=MODEL_DISPLAY_NAME,
replica_count=REPLICA_COUNT,
machine_type=MACHINE_TYPE,
accelerator_type=ACCELERATOR_TYPE,
accelerator_count=ACCELERATOR_COUNT,
training_fraction_split=TRAINING_FRACTION_SPLIT,
validation_fraction_split=VALIDATION_FRACTION_SPLIT,
test_fraction_split=TEST_FRACTION_SPLIT,
region=REGION,
project_id=PROJECT_ID,
)
创建模型¶
模型是一个可以容纳多个模型版本的容器。可以通过 MLEngineCreateModelOperator 创建新模型。应使用包含有关模型信息的字典定义 model 字段。此字典中 name 是必需字段。
此操作已弃用。模型是运行创建任何类型的训练作业的 Vertex AI 操作的结果。例如，您可以使用 CreateCustomPythonPackageTrainingJobOperator。运行此操作的结果将是保存在模型注册表中的可随时使用的模型。
create_custom_python_package_training_job = CreateCustomPythonPackageTrainingJobOperator(
task_id="create_custom_python_package_training_job",
staging_bucket=f"gs://{CUSTOM_PYTHON_GCS_BUCKET_NAME}",
display_name=PACKAGE_DISPLAY_NAME,
python_package_gcs_uri=PYTHON_PACKAGE_GCS_URI,
python_module_name=PYTHON_MODULE_NAME,
container_uri=TRAIN_IMAGE,
model_serving_container_image_uri=DEPLOY_IMAGE,
bigquery_destination=f"bq://{PROJECT_ID}",
# run params
dataset_id=tabular_dataset_id,
model_display_name=MODEL_DISPLAY_NAME,
replica_count=REPLICA_COUNT,
machine_type=MACHINE_TYPE,
accelerator_type=ACCELERATOR_TYPE,
accelerator_count=ACCELERATOR_COUNT,
training_fraction_split=TRAINING_FRACTION_SPLIT,
validation_fraction_split=VALIDATION_FRACTION_SPLIT,
test_fraction_split=TEST_FRACTION_SPLIT,
region=REGION,
project_id=PROJECT_ID,
)
获取模型¶
可以使用 MLEngineGetModelOperator 获取之前创建的模型。要获取正确的模型，必须在操作符中定义 model_name。
此操作符已弃用。请改用 GetModelOperator。
get_model = GetModelOperator(
task_id="get_model", region=REGION, project_id=PROJECT_ID, model_id=model_id_v1
)
可以使用 Jinja 模板 和 project_id 和 model 字段动态确定其值。结果保存到 XCom 中，以便其他操作符使用它们。在这种情况下，BashOperator 用于打印模型信息。
get_model_result = BashOperator(
bash_command=f"echo {get_model.output}",
task_id="get_model_result",
)
创建模型版本¶
模型版本是模型容器的一个子集，其中代码运行。可以通过 MLEngineCreateVersionOperator 创建模型的新版本。必须通过 model_name 指定模型，version 参数应包含有关该版本的所有信息。在 version 参数的字典中，name 字段是必需的。
此运算符已弃用。请改用 CreateCustomPythonPackageTrainingJobOperator。在这种情况下，可以通过在运行训练作业时在 parent_model 参数中指定现有模型 ID 来创建特定模型的新版本。这将确保训练新版本的模型，而不是创建新模型。
create_custom_python_package_training_job = CreateCustomPythonPackageTrainingJobOperator(
task_id="create_custom_python_package_training_job",
staging_bucket=f"gs://{CUSTOM_PYTHON_GCS_BUCKET_NAME}",
display_name=PACKAGE_DISPLAY_NAME,
python_package_gcs_uri=PYTHON_PACKAGE_GCS_URI,
python_module_name=PYTHON_MODULE_NAME,
container_uri=TRAIN_IMAGE,
model_serving_container_image_uri=DEPLOY_IMAGE,
bigquery_destination=f"bq://{PROJECT_ID}",
# run params
dataset_id=tabular_dataset_id,
model_display_name=MODEL_DISPLAY_NAME,
replica_count=REPLICA_COUNT,
machine_type=MACHINE_TYPE,
accelerator_type=ACCELERATOR_TYPE,
accelerator_count=ACCELERATOR_COUNT,
training_fraction_split=TRAINING_FRACTION_SPLIT,
validation_fraction_split=VALIDATION_FRACTION_SPLIT,
test_fraction_split=TEST_FRACTION_SPLIT,
region=REGION,
project_id=PROJECT_ID,
)
还可以使用 CreateCustomPythonPackageTrainingJobOperator 创建具有不同参数的更多版本。
create_custom_python_package_training_job_v2 = CreateCustomPythonPackageTrainingJobOperator(
task_id="create_custom_python_package_training_job_v2",
staging_bucket=f"gs://{CUSTOM_PYTHON_GCS_BUCKET_NAME}",
display_name=PACKAGE_DISPLAY_NAME,
python_package_gcs_uri=PYTHON_PACKAGE_GCS_URI,
python_module_name=PYTHON_MODULE_NAME,
container_uri=TRAIN_IMAGE,
model_serving_container_image_uri=DEPLOY_IMAGE,
bigquery_destination=f"bq://{PROJECT_ID}",
parent_model=model_id_v1,
# run params
dataset_id=tabular_dataset_id,
model_display_name=MODEL_DISPLAY_NAME,
replica_count=REPLICA_COUNT,
machine_type=MACHINE_TYPE,
accelerator_type=ACCELERATOR_TYPE,
accelerator_count=ACCELERATOR_COUNT,
training_fraction_split=TRAINING_FRACTION_SPLIT,
validation_fraction_split=VALIDATION_FRACTION_SPLIT,
test_fraction_split=TEST_FRACTION_SPLIT,
region=REGION,
project_id=PROJECT_ID,
)
管理模型版本¶
默认情况下，模型代码将使用默认模型版本运行。您可以通过 MLEngineSetDefaultVersionOperator 设置模型版本，方法是指定 model_name 和 version_name 参数。
此运算符已弃用。请改用 SetDefaultVersionOnModelOperator。可以将要设置为默认值的所需模型版本与 model_id 参数中的模型 ID 一起传递，格式为 projects/{project}/locations/{location}/models/{model_id}@{version_id} 或 projects/{project}/locations/{location}/models/{model_id}@{version_alias}。默认情况下，创建的第一个模型版本将标记为默认版本。
set_default_version = SetDefaultVersionOnModelOperator(
task_id="set_default_version",
project_id=PROJECT_ID,
region=REGION,
model_id=model_id_v2,
)
要列出可用的模型版本，请使用 MLEngineListVersionsOperator，同时指定 model_name 参数。
此操作符已弃用。请改用 ListModelVersionsOperator。您可以在 model_id 参数中传递所需模型的名称。如果使用版本别名传递模型 ID，则该操作符将输出此模型可用的所有版本。
list_model_versions = ListModelVersionsOperator(
task_id="list_model_versions", region=REGION, project_id=PROJECT_ID, model_id=model_id_v2
)
进行预测¶
可以使用 MLEngineStartBatchPredictionJobOperator 启动 Google Cloud AI Platform 预测作业。要指定模型来源，您需要提供 model_name、uri 或 model_name 和 version_name。如果您未提供 version_name，该操作符将使用默认模型版本。
此操作符已弃用。请改用 CreateBatchPredictionJobOperator。
create_batch_prediction_job = CreateBatchPredictionJobOperator(
task_id="create_batch_prediction_job",
job_display_name=JOB_DISPLAY_NAME,
model_name=model_id_v2,
predictions_format="bigquery",
bigquery_source=BQ_SOURCE,
bigquery_destination_prefix=f"bq://{PROJECT_ID}",
region=REGION,
project_id=PROJECT_ID,
machine_type=MACHINE_TYPE,
)
清理¶
可以使用 MLEngineDeleteVersionOperator 通过 version_name 和 model_name 参数删除模型版本。
此运算符已弃用。请改用 DeleteModelVersionOperator。无法在模型上删除默认版本。
delete_model_version_1 = DeleteModelVersionOperator(
task_id="delete_model_version_1",
project_id=PROJECT_ID,
region=REGION,
model_id=model_id_v1,
trigger_rule=TriggerRule.ALL_DONE,
)
您还可以通过提供 model_name 参数，使用 MLEngineDeleteModelOperator 删除模型。
此运算符已弃用。请改用 DeleteModelOperator。
delete_model = DeleteModelOperator(
task_id="delete_model",
project_id=PROJECT_ID,
region=REGION,
model_id=model_id_v2,
trigger_rule=TriggerRule.ALL_DONE,
)
评估模型¶
此函数已弃用。旧版 MLEngine 和新功能的所有功能均可在 Vertex AI 平台上使用。要创建和查看模型评估，请查看文档：使用 Vertex AI 评估模型