Merge branch 'main' into ml_ops_3v4

notebooks/community/ml_ops/stage3/mlops_formalization.ipynb

@@ -764,12 +764,15 @@
       "source": [
         "import json\n",
         "\n",
-        "with tf.io.gfile.GFile(\n",
-        "    \"gs://\" + dataset.labels[\"user_metadata\"] + \"/metadata.jsonl\", \"r\"\n",
-        ") as f:\n",
-        "    metadata = json.load(f)\n",
+        "try:\n",
+        "    with tf.io.gfile.GFile(\n",
+        "        \"gs://\" + dataset.labels[\"user_metadata\"] + \"/metadata.jsonl\", \"r\"\n",
+        "    ) as f:\n",
+        "        metadata = json.load(f)\n",
         "\n",
-        "print(metadata)"
+        "    print(metadata)\n",
+        "except:\n",
+        "    print(\"no metadata\")"
       ]
     },
     {
@@ -1132,6 +1135,9 @@
         "\n",
         "- Query the BigQuery table for all examples (parse_bq_record).\n",
         "- Split the examples into training, evaluation and test datasets (split_dataset).\n",
+        "\n",
+        "#### Data preprocessing\n",
+        "\n",
         "- Preprocess each example (preprocessing_fn).\n",
         "- Write the preprocessed data to a Cloud Storage bucket as TFRecords.\n",
         "- Write the transformation function artifacts to a Cloud Storage bucket.\n",
@@ -1813,13 +1819,15 @@
         "\n",
         "### Build model architecture task\n",
         "\n",
-        "BLAH\n",
+        "- Construct the model architecture for the base model and save as a Vertex AI Model resource.\n",
         "\n",
         "### Triggers\n",
         "\n",
         "Within the CI/CD process, the model pipeline is triggered for one or more of the following example reasons, while not exhaustive:\n",
         "\n",
-        "- BLAH"
+        "- If the data pipeline is re-executed.\n",
+        "- Code changes to the build model architecture task.\n",
+        "- The model architecture is being replaced."
       ]
     },
     {
@@ -1837,7 +1845,6 @@
         "- Use the metadata information on the feature types to build the input layer.\n",
         "- Build the DNN body of the model.\n",
         "- Save the base model artifacts to the Cloud Storage location.\n",
-        "- Create a Vertex AI Model resource for the base model.\n",
         "\n",
         "The component returns the full resource name of the generated Vertex AI Model resource."
       ]
@@ -1988,7 +1995,8 @@
         "\n",
         "Next, construct the pipeline with the following tasks:\n",
         "\n",
-        "- Build the base model architecture."
+        "- Build the base model architecture.\n",
+        "- Create a Vertex AI Model resource for the base model."
       ]
     },
     {
@@ -2007,7 +2015,7 @@
         "    bucket: str,\n",
         "    project: str = PROJECT_ID,\n",
         "    region: str = REGION,\n",
-        "    labels: list = [{\"base_model\": 1}],\n",
+        "    labels: dict = {\"base_model\": 1},\n",
         "):\n",
         "    from google_cloud_pipeline_components import aiplatform as gcc_aip\n",
         "\n",
@@ -2085,7 +2093,7 @@
         "id": "view_pipleline_results:model_build"
       },
       "source": [
-        "### View BLAH"
+        "### View the model pipeline execution results"
       ]
     },
     {
@@ -2112,7 +2120,32 @@
     {
       "cell_type": "markdown",
       "metadata": {
-        "id": "construct_training_package"
+        "id": "formalize_training_pipeline_intro"
+      },
+      "source": [
+        "## Formalizing training pipeline introduction\n",
+        "\n",
+        "The training pipeline consists of training the model.\n",
+        "\n",
+        "### Train the model task\n",
+        "\n",
+        "- Retrieve the model architecture.\n",
+        "- ?? Hypertune BLAH\n",
+        "- Train the model\n",
+        "- Evaluate the model\n",
+        "\n",
+        "### Triggers\n",
+        "\n",
+        "Within the CI/CD process, the training pipeline is triggered for one or more of the following example reasons, while not exhaustive:\n",
+        "\n",
+        "- If the data pipeline is re-executed.\n",
+        "- If the model pipeline is re-executed.\n",
+        "- Code changes to the model training task."
+      ]
+    },
+    {
+      "cell_type": "markdown",
+      "metadata": {
       },
       "source": [
         "### Construct the training package\n",
@@ -2552,7 +2585,7 @@
         "    evalparams = {}\n",
         "    evalparams[\"batch_size\"] = args.batch_size\n",
         "    metrics = train.evaluate(model, evalparams, test_data_file_pattern, label_column, transform_feature_spec)\n",
-        "    with tf.io.gfile.GFile(os.path.join(args.model_dir, \"metrics.txt\", \"w\")) as f:\n",
+        "    with tf.io.gfile.GFile(os.path.join(args.model_dir, \"metrics.txt\"), \"w\") as f:\n",
         "        f.write(str(metrics))\n",
         "\n",
         "get_data()\n",
@@ -2570,7 +2603,7 @@
     {
       "cell_type": "markdown",
       "metadata": {
-        "id": "test_package_locally"
+        "id": "test_package_locally:id"
       },
       "source": [
         "### Test training package locally\n",
@@ -2582,15 +2615,40 @@
       "cell_type": "code",
       "execution_count": null,
       "metadata": {
-        "id": "test_package_locally"
+        "id": "test_package_locally:id"
       },
       "outputs": [],
       "source": [
-        "DATASET_ID = dataset.resource_name\n",
-        "MODEL_ID = vertex_custom_model.resource_name\n",
+        "DATASET_ID = dataset_id\n",
+        "MODEL_ID = model_id\n",
         "!cd custom; python3 -m trainer.task --model-id={MODEL_ID} --dataset-id={DATASET_ID} --experiment='chicago' --run='test' --project={PROJECT_ID} --epochs=5 --model-dir=/tmp --evaluate=True"
       ]
     },
+    {
+      "cell_type": "markdown",
+      "metadata": {
+        "id": "tarball_training_script"
+      },
+      "source": [
+        "#### Store training script on your Cloud Storage bucket\n",
+        "\n",
+        "Next, you package the training folder into a compressed tar ball, and then store it in your Cloud Storage bucket."
+      ]
+    },
+    {
+      "cell_type": "code",
+      "execution_count": null,
+      "metadata": {
+        "id": "tarball_training_script"
+      },
+      "outputs": [],
+      "source": [
+        "! rm -f custom.tar custom.tar.gz\n",
+        "! tar cvf custom.tar custom\n",
+        "! gzip custom.tar\n",
+        "! gsutil cp custom.tar.gz $BUCKET_NAME/trainer_chicago.tar.gz"
+      ]
+    },
     {
       "cell_type": "markdown",
       "metadata": {
@@ -2647,19 +2705,182 @@
         "    description=\"Custom tabular binary classification training\",\n",
         ")\n",
         "def pipeline(\n",
-        "    import_file: str,\n",
         "    display_name: str,\n",
+        "    dataset_id: str,\n",
+        "    model_id: str,\n",
         "    python_package: str,\n",
         "    python_module: str,\n",
+        "    args: str,\n",
+        "    container_uri: str,\n",
+        "    model_serving_container_image_uri: str,\n",
+        "    machine_type: str,\n",
+        "    bucket: str,\n",
+        "    replica_count: int = 1,\n",
+        "    accelerator_type: str = None,\n",
+        "    accelerator_count: int = 0,\n",
         "    project: str = PROJECT_ID,\n",
         "    region: str = REGION,\n",
         "):\n",
         "\n",
-        "    dataset_op = gcc_aip.TabularDatasetCreateOp(\n",
-        "        project=project, display_name=display_name, bq_source=import_file\n",
+        "    training_op = gcc_aip.CustomPythonPackageTrainingJobRunOp(\n",
+        "        project=project,\n",
+        "        display_name=display_name,\n",
+        "        # Training\n",
+        "        python_package_gcs_uri=python_package,\n",
+        "        python_module_name=python_module,\n",
+        "        container_uri=container_uri,\n",
+        "        staging_bucket=bucket,\n",
+        "        # annotation_schema_uri=aip.schema.dataset.annotation.tabular.classification,\n",
+        "        args=args,\n",
+        "        replica_count=replica_count,\n",
+        "        machine_type=machine_type,\n",
+        "        accelerator_type=accelerator_type,\n",
+        "        accelerator_count=accelerator_count,\n",
+        "        # Serving - As part of this operation, the model is registered to Vertex AI\n",
+        "        model_serving_container_image_uri=model_serving_container_image_uri,\n",
+        "        model_display_name=display_name,\n",
+        "    )\n",
+        "\n",
+        "    endpoint_op = gcc_aip.EndpointCreateOp(\n",
+        "        project=project,\n",
+        "        location=region,\n",
+        "        display_name=display_name,\n",
+        "    ).after(training_op)\n",
+        "\n",
+        "    deploy_op = gcc_aip.ModelDeployOp(\n",
+        "        model=training_op.outputs[\"model\"],\n",
+        "        endpoint=endpoint_op.outputs[\"endpoint\"],\n",
+        "        dedicated_resources_min_replica_count=1,\n",
+        "        dedicated_resources_max_replica_count=1,\n",
+        "        dedicated_resources_machine_type=\"n1-standard-4\",\n",
         "    )"
       ]
     },
+    {
+      "cell_type": "markdown",
+      "metadata": {
+        "id": "run_pipeline:model_train"
+      },
+      "source": [
+        "### Compile and execute the model training pipeline\n",
+        "\n",
+        "Next, you compile the pipeline and then execute it. The pipeline takes the following parameters, which are passed as the dictionary `parameter_values`:\n",
+        "\n",
+        "- `dataset_id`: The full resource name of the corresponding Vertex AI Dataset.\n",
+        "- `model_id`: The full resource name of the corresponding Vertex AI Model architecture.\n",
+        "- `display_name`: The display name for the trained Vertex AI Model resource.\n",
+        "- `python_package`: The Python package for the custom training job.\n",
+        "- `python_module`: The Python module in the package to execute.\n",
+        "- `args`: The command line arguments to pass to the Python module.\n",
+        "- `container_uri`: The training container image.\n",
+        "- `model_serving_container_image_uri`: The associated deployment container image.\n",
+        "- `machine_type`: The VM for executing the training job.\n",
+        "- `replica_count`: The number of virtual machines -- if doing distributed multi-machine training.\n",
+        "- `accelerator_type`: The type of HW accelerators -- if any.\n",
+        "- `accelerator_count`: The number of HW accelerators -- if any.\n",
+        "- `bucket`: The Cloud Storage location to store the model artifacts.\n",
+        "- `project`: The project ID.\n",
+        "- `region`: The region."
+      ]
+    },
+    {
+      "cell_type": "code",
+      "execution_count": null,
+      "metadata": {
+        "id": "run_pipeline:model_train"
+      },
+      "outputs": [],
+      "source": [
+        "PIPELINE_ROOT = \"{}/pipeline_root/model-train\".format(BUCKET_NAME)\n",
+        "\n",
+        "compiler.Compiler().compile(pipeline_func=pipeline, package_path=\"model_train.json\")\n",
+        "\n",
+        "pipeline = aip.PipelineJob(\n",
+        "    display_name=\"model-train\",\n",
+        "    template_path=\"model_train.json\",\n",
+        "    pipeline_root=PIPELINE_ROOT,\n",
+        "    parameter_values={\n",
+        "        \"dataset_id\": dataset_id,\n",
+        "        \"model_id\": model_id,\n",
+        "        \"display_name\": \"chicago\" + TIMESTAMP,\n",
+        "        \"python_package\": f\"{BUCKET_NAME}/trainer_chicago.tar.gz\",\n",
+        "        \"python_module\": \"trainer.task\",\n",
+        "        \"args\": [\n",
+        "            \"--dataset-id\",\n",
+        "            dataset_id,\n",
+        "            \"--model-id\",\n",
+        "            model_id,\n",
+        "            \"--experiment\",\n",
+        "            \"chicago\",\n",
+        "            \"--run\",\n",
+        "            \"retrain\",\n",
+        "            \"--epochs\",\n",
+        "            \"50\",\n",
+        "            \"--evaluate\",\n",
+        "            \"True\",\n",
+        "            \"--project\",\n",
+        "            PROJECT_ID,\n",
+        "        ],\n",
+        "        \"container_uri\": TRAIN_IMAGE,\n",
+        "        \"model_serving_container_image_uri\": DEPLOY_IMAGE,\n",
+        "        \"machine_type\": TRAIN_COMPUTE,\n",
+        "        \"replica_count\": 1,\n",
+        "        \"accelerator_type\": TRAIN_GPU.name,\n",
+        "        \"accelerator_count\": TRAIN_NGPU,\n",
+        "        \"bucket\": BUCKET_NAME,\n",
+        "        \"project\": PROJECT_ID,\n",
+        "        \"region\": REGION,\n",
+        "    },\n",
+        ")\n",
+        "\n",
+        "pipeline.run()\n",
+        "\n",
+        "! rm -rf model_train.json custom_tar.gz custom"
+      ]
+    },
+    {
+      "cell_type": "markdown",
+      "metadata": {
+        "id": "view_pipleline_results:model_train"
+      },
+      "source": [
+        "### View the training pipeline results"
+      ]
+    },
+    {
+      "cell_type": "code",
+      "execution_count": null,
+      "metadata": {
+        "id": "view_pipleline_results:model_train"
+      },
+      "outputs": [],
+      "source": [
+        "print(\"custompythonpackagetrainingjob-run\")\n",
+        "artifacts = print_pipeline_output(pipeline, 'custompythonpackagetrainingjob-run')\n",
+        "print('\\n')print(\"custompythonpackagetrainingjob-run\")\n",
+        "artifacts = print_pipeline_output(pipeline, 'custompythonpackagetrainingjob-run')\n",
+        "print('\\n')\n",
+        "output = !gsutil cat $artifacts\n",
+        "output = json.loads(output[0])\n",
+        "model_id = output['artifacts']['model']['artifacts'][0]['metadata']['resourceName']\n",
+        "print('\\n')\n",
+        "print(model_id)\n",
+        "print('\\n')\n",
+        "\n",
+        "print(\"endpoint-create\")\n",
+        "artifacts = print_pipeline_output(pipeline, 'endpoint-create')\n",
+        "print('\\n')\n",
+        "print(\"model-deploy\")\n",
+        "artifacts = print_pipeline_output(pipeline, 'model-deploy')\n",
+        "print('\\n')\n",
+        "print(\"endpoint-create\")\n",
+        "artifacts = print_pipeline_output(pipeline, 'endpoint-create')\n",
+        "print('\\n')\n",
+        "print(\"model-deploy\")\n",
+        "artifacts = print_pipeline_output(pipeline, 'model-deploy')\n",
+        "print('\\n')"
+      ]
+    },
     {
       "cell_type": "code",
       "execution_count": null,
@@ -2691,17 +2912,6 @@
         "    if \"BUCKET_NAME\" in globals():\n",
         "        ! gsutil rm -r $BUCKET_NAME"
       ]
-    },
-    {
-      "cell_type": "code",
-      "execution_count": null,
-      "metadata": {
-        "id": "81ec305f286d"
-      },
-      "outputs": [],
-      "source": [
-        "! gsutil rm -r $BUCKET_NAME"
-      ]
     }
   ],
   "metadata": {