fix: Fixed kernel spec by forcing it to use python3

2026-09-27 15:42:05 +00:00 · 2022-08-19 17:51:06 -04:00
4 changed files with 1069 additions and 351 deletions
@@ -1214,20 +1214,6 @@
        "     [1.0,3.0,\"cat1\"],\n",
        "     [2.0,4.0,\"cat2\"]\n",
        "    ]}\n",
-        "    \n",
-        "**BigQuery**\n",
-        "\n",
-        "Each row is converted to a JSON array. For example:\n",
-        "\n",
-        "    [1.0,3.0,\"cat1\"]\n",
-        "    [2.0,4.0,\"cat2\"]\n",
-        "    \n",
-        "The batch server generates the pivot data with the same format. The generated pivot data is then wrapped into a payload request:\n",
-        "\n",
-        "    {\"instances\": [\n",
-        "     [1.0,3.0,\"cat1\"],\n",
-        "     [2.0,4.0,\"cat2\"]\n",
-        "    ]}\n",
        "\n",
        "**TFRecords**\n",
        "\n",
@@ -55,7 +55,7 @@
        "    </a>\n",
        "  </td>\n",
        "  <td>\n",
-        "    <a href=\"https://console.cloud.google.com/vertex-ai/workbench/deploy-notebook?download_url=https://raw.githubusercontent.com/GoogleCloudPlatform/vertex-ai-samples/main/notebooks/official/experiments/comparing_pipeline_runs.ipynb\">\n",
+        "    <a href=\"https://console.cloud.google.com/vertex-ai/workbench/deploy-notebook?download_url=https://github.com/GoogleCloudPlatform/vertex-ai-samples/blob/main/notebooks/official/experiments/comparing_pipeline_runs.ipynb\">\n",
        "      <img src=\"https://lh3.googleusercontent.com/UiNooY4LUgW_oTvpsNhPpQzsstV5W8F7rYgxgGBD85cWJoLmrOzhVs_ksK_vgx40SHs7jCqkTkCk=e14-rj-sc0xffffff-h130-w32\" alt=\"Vertex AI logo\">\n",
        "      Open in Vertex AI Workbench\n",
        "    </a>\n",
@@ -70,8 +70,6 @@
        "id": "tvgnzT1CKxrO"
      },
      "source": [
-        "# Compare pipeline runs with Vertex AI Experiments\n",
-        "\n",
        "## Overview\n",
        "\n",
        "Depending on the model life cycle of your data science team, you would like to experiment and track training Pipeline runs and its associated parameters. Then, you would to compare runs of these Pipelines to each others in order to figure out which is the best configuration generates the model you will register in the Vertex AI Model Registry."
@@ -209,8 +207,10 @@
        "if IS_WORKBENCH_NOTEBOOK:\n",
        "    USER_FLAG = \"--user\"\n",
        "\n",
-        "!pip3 install {USER_FLAG} --force-reinstall 'google-cloud-aiplatform>=1.15' -q --no-warn-conflicts\n",
-        "!pip3 install {USER_FLAG} kfp -q --no-warn-conflicts"
+        "! pip install {USER_FLAG} --upgrade google-cloud-aiplatform \\\n",
+        "                                    google-cloud-storage \\\n",
+        "                                    google-auth \\\n",
+        "                                    kfp -q"
      ]
    },
    {
@@ -270,7 +270,7 @@
        "\n",
        "1. [Make sure that billing is enabled for your project](https://cloud.google.com/billing/docs/how-to/modify-project).\n",
        "\n",
-        "1. [Enable APIs](https://console.cloud.google.com/flows/enableapi?apiid=cloudresourcemanager.googleapis.com,aiplatform.googleapis.com).\n",
+        "1. [Enable the Vertex AI API](https://console.cloud.google.com/flows/enableapi?apiid=aiplatform.googleapis.com).\n",
        "\n",
        "1. If you are running this notebook locally, you will need to install the [Cloud SDK](https://cloud.google.com/sdk).\n",
        "\n",
@@ -451,14 +451,9 @@
        "3. In the **Service account name** field, enter a name, and\n",
        "   click **Create**.\n",
        "\n",
-        "4. In the **Grant this service account access to project** section, click the **Role** drop-down list. Type and select\n",
-        "the following role into the filter box:\n",
-        "\n",
-        "    *   Storage Admin\n",
-        "    *   Storage Object Admin\n",
-        "    *   Service Account User\n",
-        "    *   Vertex AI Administrator\n",
-        "\n",
+        "4. In the **Grant this service account access to project** section, click the **Role** drop-down list. Type \"Vertex AI\"\n",
+        "into the filter box, and select\n",
+        "   **Vertex AI Administrator**. Type \"Storage Object Admin\" into the filter box, and select **Storage Object Admin**.\n",
        "\n",
        "5. Click *Create*. A JSON file that contains your key downloads to your\n",
        "local environment.\n",
@@ -725,7 +720,6 @@
        "import kfp.v2.dsl as dsl\n",
        "# Vertex AI\n",
        "from google.cloud import aiplatform as vertex_ai\n",
-        "from google.cloud.aiplatform_v1.types.pipeline_state import PipelineState\n",
        "from kfp.v2.dsl import Metrics, Model, Output, component"
      ]
    },
@@ -744,7 +738,6 @@
        "EXPERIMENT_NAME = f\"{PROJECT_ID}-{TASK}-{MODEL_TYPE}-{UUID}\"\n",
        "\n",
        "# Pipeline\n",
-        "PIPELINE_TEMPLATE_FILE = \"pipeline.json\"\n",
        "PIPELINE_URI = f\"{BUCKET_URI}/pipelines\"\n",
        "TRAIN_URI = f\"{BUCKET_URI}/iris/iris_data.csv\"\n",
        "LABEL_URI = f\"{BUCKET_URI}/iris/iris_target.csv\"\n",
@@ -826,7 +819,9 @@
      "source": [
        "Before you start running your pipeline experiments, you have to formalize your training as pipeline component.\n",
        "\n",
-        "To do that, you build the pipeline by using the `kfp.v2.dsl.component` decorator to convert your training task into a pipeline component. "
+        "To do that, you will use the `kfp.v2.dsl.component` decorator to convert your training task into a pipeline component.\n",
+        "\n",
+        "Training code will import required libraries to train,evaluate and save a model with mentioned features. "
      ]
    },
    {
@@ -1080,7 +1075,7 @@
        "\n",
        "    job = vertex_ai.PipelineJob(\n",
        "        display_name=f\"{EXPERIMENT_NAME}-pipeline-run-{i}\",\n",
-        "        template_path=PIPELINE_TEMPLATE_FILE,\n",
+        "        template_path=\"pipeline.json\",\n",
        "        pipeline_root=PIPELINE_URI,\n",
        "        parameter_values={\n",
        "            \"train_uri\": TRAIN_URI,\n",
@@ -1165,9 +1160,10 @@
      "source": [
        "# Get the PipelineJob resource using the experiment run name\n",
        "pipeline_experiments_df = vertex_ai.get_experiment_df(EXPERIMENT_NAME)\n",
-        "job = vertex_ai.PipelineJob.get(pipeline_experiments_df.run_name[0])\n",
-        "print(\"Pipeline job name: \", job.resource_name)\n",
-        "print(\"Pipeline Run UI link: \", job._dashboard_uri())"
+        "for i in range(5):\n",
+        "    job = vertex_ai.PipelineJob.get(pipeline_experiments_df.run_name[i])\n",
+        "    print(job.resource_name)\n",
+        "    print(job._dashboard_uri())"
      ]
    },
    {
@@ -1195,16 +1191,13 @@
      "outputs": [],
      "source": [
        "# Delete the pipeline\n",
-        "while True:\n",
-        "    for i in range(0, len(runs)):\n",
-        "        pipeline_job = vertex_ai.PipelineJob.get(pipeline_experiments_df.run_name[i])\n",
-        "        if pipeline_job.state != PipelineState.PIPELINE_STATE_SUCCEEDED:\n",
-        "            print(\"Pipeline job is still running...\")\n",
-        "            time.sleep(60)\n",
-        "        else:\n",
-        "            print(\"Pipeline job is complete.\")\n",
-        "            pipeline_job.delete()\n",
-        "    break\n",
+        "# Get the PipelineJob resource using the experiment run name\n",
+        "pipeline_experiments_df = vertex_ai.get_experiment_df(EXPERIMENT_NAME)\n",
+        "for i in range(5):\n",
+        "    job = vertex_ai.PipelineJob.get(pipeline_experiments_df.run_name[i])\n",
+        "    print(job.resource_name)\n",
+        "    print(job._dashboard_uri())\n",
+        "    job.delete()\n",
        "\n",
        "# Delete experiment\n",
        "exp = vertex_ai.Experiment(EXPERIMENT_NAME)\n",
@@ -1213,11 +1206,7 @@
        "# Delete bucket\n",
        "delete_bucket = False\n",
        "if delete_bucket or os.getenv(\"IS_TESTING\"):\n",
-        "    ! gsutil rm -rf {BUCKET_URI}\n",
-        "\n",
-        "# Remove local files\n",
-        "\n",
-        "!rm {PIPELINE_TEMPLATE_FILE}"
+        "    ! gsutil rm -rf {BUCKET_URI}"
      ]
    }
  ],
@@ -425,19 +425,9 @@
        "3. In the **Service account name** field, enter a name, and\n",
        "   click **Create**.\n",
        "\n",
-        "4. In the **Grant this service account access to project** section, click the **Role** drop-down list. Type the following role and select them\n",
-        "\n",
-        "      - Artifact Registry Administrator\n",
-        "      - Artifact Registry Repository Administrator\n",
-        "      - Cloud Build Editor\n",
-        "      - Compute Network Admin\n",
-        "      - Dataproc Administrator\n",
-        "      - Dataproc Worker\n",
-        "      - Service Account User\n",
-        "      - Storage Admin\n",
-        "      - Storage Object Admin\n",
-        "      - Vertex AI Administrator\n",
-        "\n",
+        "4. In the **Grant this service account access to project** section, click the **Role** drop-down list. Type \"Vertex AI\"\n",
+        "into the filter box, and select\n",
+        "   **Vertex AI Administrator**. Type \"Storage Object Admin\" into the filter box, and select **Storage Object Admin**.\n",
        "\n",
        "5. Click *Create*. A JSON file that contains your key downloads to your\n",
        "local environment.\n",
@@ -649,10 +639,15 @@
      },
      "outputs": [],
      "source": [
+        "from pathlib import Path as path\n",
+        "\n",
+        "DATA_PATH = path(\"content/path/\")\n",
        "PUBLIC_DATA_URI = \"gs://cloud-samples-data/vertex-ai/dataset-management/datasets/loan_eligibilty/data.csv\"\n",
        "FEATURES_TRAIN_URI = f\"{BUCKET_URI}/data/features/snapshots/{UUID}\"\n",
        "\n",
-        "!gsutil cp -r $PUBLIC_DATA_URI $FEATURES_TRAIN_URI"
+        "!mkdir -m 777 -p $DATA_PATH\n",
+        "!gsutil cp -r $PUBLIC_DATA_URI $DATA_PATH\n",
+        "!gsutil cp -r $DATA_PATH $FEATURES_TRAIN_URI"
      ]
    },
    {
@@ -738,7 +733,9 @@
        "from google_cloud_pipeline_components import aiplatform as vertex_ai_components\n",
        "from kfp.v2 import compiler, dsl\n",
        "from kfp.v2.dsl import (Artifact, ClassificationMetrics, Condition, Input,\n",
-        "                        Metrics, Output, component)"
+        "                        Metrics, Output, component)\n",
+        "\n",
+        "ID = random.randint(1, 10000)"
      ]
    },
    {
@@ -770,7 +767,7 @@
        "MODEL_NAME = f\"{ML_APPLICATION}-{TASK}-{MODEL_TYPE}-{VERSION}\"\n",
        "\n",
        "# Preprocessing\n",
-        "PREPROCESSING_BATCH_ID = f\"data-preprocessing-{UUID}\"\n",
+        "PREPROCESSING_BATCH_ID = f\"data-preprocessing-{ID}\"\n",
        "PREPROCESSING_PYTHON_FILE_URI = f\"{BUCKET_URI}/src/data_preprocessing.py\"\n",
        "PROCESSED_DATA_URI = f\"{BUCKET_URI}/data/processed\"\n",
        "PREPROCESSING_ARGS = [\n",
@@ -781,11 +778,11 @@
        "]\n",
        "\n",
        "# Dataset\n",
-        "DATASET_NAME = f\"preprocessed-dataset-{UUID}\"\n",
+        "DATASET_NAME = f\"preprocessed-dataset-{ID}\"\n",
        "GCS_PREPROCESSED_URI = f\"{PROCESSED_DATA_URI}/*/?.csv\"\n",
        "\n",
        "# Training\n",
-        "TRAINING_BATCH_ID = f\"model-training-{UUID}\"\n",
+        "TRAINING_BATCH_ID = f\"model-training-{ID}\"\n",
        "TRAINING_PYTHON_FILE_URI = f\"{BUCKET_URI}/src/model_training.py\"\n",
        "MODEL_URI = f\"{BUCKET_URI}/deliverables/model/rfor/{UUID}/train_model\"\n",
        "METRICS_URI = f\"{BUCKET_URI}/deliverables/metrics/rfor/{UUID}/train_metrics.json\"\n",
@@ -803,7 +800,7 @@
        "AUPR_HYPERTUNE_CONDITION = \"[AUPR_HYPERTUNE]\"\n",
        "\n",
        "# Hypertuning\n",
-        "HPT_TRAINING_BATCH_ID = f\"hyper-tuning-{UUID}\"\n",
+        "HPT_TRAINING_BATCH_ID = f\"hyper-tuning-{ID}\"\n",
        "HPT_PYTHON_FILE_URI = f\"{BUCKET_URI}/src/hp_tuning.py\"\n",
        "HPT_MODEL_URI = f\"{BUCKET_URI}/deliverables/model/rfor/{UUID}/model\"\n",
        "HPT_METRICS_URI = f\"{BUCKET_URI}/deliverables/metrics/rfor/{UUID}/metrics.json\"\n",
@@ -912,7 +909,6 @@
        "\n",
        "\"\"\"\n",
        "data_preprocessing.py is the module for\n",
-        "\n",
        "  - ingest data\n",
        "  - do simple preprocessing tasks\n",
        "  - upload processed data to gcs\n",
@@ -1025,7 +1021,8 @@
        "\n",
        "    spark = (SparkSession.builder\n",
        "             .master(\"local[*]\")\n",
-        "             .appName(\"loan eligibility\")\n",
+        "             .appName(\"spark go live\")\n",
+        "             .config('spark.ui.port', '4050')\n",
        "             .getOrCreate())\n",
        "    try:\n",
        "        logger.info(f'spark version: {spark.sparkContext.version}')\n",
@@ -1044,7 +1041,12 @@
        "        training_data_raw_df.show(truncate=False)\n",
        "\n",
        "        logger.info(f'load prepared data to {output_data_path}.')\n",
-        "        training_data_raw_df.write.mode('overwrite').csv(str(output_data_path), header=True)\n",
+        "        if output_data_path.startswith('gs://'):\n",
+        "            training_data_raw_df.write.mode('overwrite').csv(str(output_data_path), header=True)\n",
+        "        else:\n",
+        "            output_file_path = Path(output_data_path)\n",
+        "            output_file_path.mkdir(parents=True, exist_ok=True)\n",
+        "            training_data_raw_df.write.mode('overwrite').csv(str(output_file_path), header=True)\n",
        "    except RuntimeError as main_error:\n",
        "        logger.error(main_error)\n",
        "    else:\n",
@@ -1374,7 +1376,8 @@
        "        logger.info('start spark session.')\n",
        "        spark = (SparkSession.builder\n",
        "                 .master(\"local[*]\")\n",
-        "                 .appName(\"loan eligibility\")\n",
+        "                 .appName(\"spark go live\")\n",
+        "                 .config('spark.ui.port', '4050')\n",
        "                 .getOrCreate())\n",
        "        logger.info(f'spark version: {spark.sparkContext.version}')\n",
        "        logger.info('start bulding pipeline.')\n",
@@ -1399,12 +1402,23 @@
        "\n",
        "        logger.info(f'load model pipeline in {model_path}.')\n",
        "        pipeline.write().overwrite().save(model_path)\n",
+        "        if model_path.startswith('gs://'):\n",
+        "            pipeline.write().overwrite().save(model_path)\n",
+        "        else:\n",
+        "            path(model_path).mkdir(parents=True, exist_ok=True)\n",
+        "            pipeline.write().overwrite().save(model_path)\n",
        "\n",
-        "        logger.info(f'Upload metrics under {metrics_path}.')     \n",
-        "        bucket = urlparse(model_path).netloc\n",
-        "        metrics_file_path = urlparse(metrics_path).path.strip('/')\n",
-        "        write_metrics(bucket, metrics, metrics_file_path)\n",
-        "        \n",
+        "        logger.info(f'Upload metrics under {metrics_path}.')\n",
+        "        if metrics_path.startswith('gs://'):\n",
+        "            bucket = urlparse(model_path).netloc\n",
+        "            metrics_file_path = urlparse(metrics_path).path.strip('/')\n",
+        "            write_metrics(bucket, metrics, metrics_file_path)\n",
+        "        else:\n",
+        "            metrics_version_path = path(metrics_path).parents[0]\n",
+        "            metrics_version_path.mkdir(parents=True, exist_ok=True)\n",
+        "            with open(metrics_path, 'w') as json_file:\n",
+        "                json.dump(metrics, json_file)\n",
+        "            json_file.close()\n",
        "    except RuntimeError as main_error:\n",
        "        logger.error(main_error)\n",
        "    else:\n",
@@ -1734,7 +1748,12 @@
        "        logger.info('start spark session.')\n",
        "        spark = (SparkSession.builder\n",
        "                 .master(\"local[*]\")\n",
-        "                 .appName(\"loan eligibility\")\n",
+        "                 .appName(\"spark go live\")\n",
+        "                 .config('spark.ui.port', '4050')\n",
+        "                 .config('spark.jars.packages', 'ml.combust.mleap:mleap-runtime_2.12:0.19.0')\n",
+        "                 .config('spark.jars.packages', 'ml.combust.mleap:mleap-base_2.12:0.19.0')\n",
+        "                 .config('spark.jars.packages', 'ml.combust.mleap:mleap-spark_2.12:0.19.0')\n",
+        "                 .config('spark.jars.packages', 'ml.combust.mleap:mleap-spark-extension_2.12:0.19.0')\n",
        "                 .getOrCreate())\n",
        "        logger.info(f'spark version: {spark.sparkContext.version}')\n",
        "        logger.info('start building pipeline.')\n",
@@ -1744,7 +1763,12 @@
        "        pipeline_cross_validator = build_hp_pipeline(preprocessing_stages, feature_engineering_stages,\n",
        "                                                     model_training_stage)\n",
        "        logger.info(f'load train data from {train_path}.')\n",
-        "        raw_data = (spark.read.format('csv')\n",
+        "        if train_path.startswith('bq://'):\n",
+        "            raw_data = spark.read.format('bigquery') \\\n",
+        "                .option('table', train_path.replace('bq://', '')) \\\n",
+        "                .load()\n",
+        "        else:\n",
+        "            raw_data = (spark.read.format('csv')\n",
        "                        .option(\"header\", \"true\")\n",
        "                        .schema(DATA_SCHEMA)\n",
        "                        .load(train_path))\n",
@@ -1757,12 +1781,23 @@
        "            print(f'{m}: {v}')\n",
        "\n",
        "        logger.info(f'load model pipeline in {model_path}.')\n",
-        "        pipeline_model.write().overwrite().save(model_path)\n",
+        "        if model_path.startswith('gs://'):\n",
+        "            pipeline_model.write().overwrite().save(model_path)\n",
+        "        else:\n",
+        "            path(model_path).mkdir(parents=True, exist_ok=True)\n",
+        "            pipeline_model.write().overwrite().save(model_path)\n",
        "\n",
        "        logger.info(f'Upload metrics under {metrics_path}.')\n",
-        "        bucket = urlparse(model_path).netloc\n",
-        "        metrics_file_path = urlparse(metrics_path).path.strip('/')\n",
-        "        write_metrics(bucket, metrics, metrics_file_path)\n",
+        "        if metrics_path.startswith('gs://'):\n",
+        "            bucket = urlparse(model_path).netloc\n",
+        "            metrics_file_path = urlparse(metrics_path).path.strip('/')\n",
+        "            write_metrics(bucket, metrics, metrics_file_path)\n",
+        "        else:\n",
+        "            metrics_version_path = path(metrics_path).parents[0]\n",
+        "            metrics_version_path.mkdir(parents=True, exist_ok=True)\n",
+        "            with open(metrics_path, 'w') as json_file:\n",
+        "                json.dump(metrics, json_file)\n",
+        "            json_file.close()\n",
        "    except RuntimeError as main_error:\n",
        "        logger.error(main_error)\n",
        "    else:\n",
@@ -1891,7 +1926,8 @@
        "      python \\\n",
        "      scikit-image \\\n",
        "      scikit-learn \\\n",
-        "      scipy  \n",
+        "      scipy \\\n",
+        "      mleap \n",
        "\n",
        "# (Required) Create the 'spark' group/user.\n",
        "# The GID and UID must be 1099. Home directory is required.\n",
@@ -2397,17 +2433,6 @@
        "if delete_bucket or os.getenv(\"IS_TESTING\"):\n",
        "    ! gsutil -m rm -r $BUCKET_URI"
      ]
-    },
-    {
-      "cell_type": "code",
-      "execution_count": null,
-      "metadata": {
-        "id": "bc7f1247f0ec"
-      },
-      "outputs": [],
-      "source": [
-        "!rm -rf $SRC $BUILD_PATH"
-      ]
    }
  ],
  "metadata": {