port: kfp2

notebooks/official/pipelines/control_flow_kfp.ipynb

@@ -129,8 +129,8 @@
       "source": [
         "! pip3 install --upgrade --quiet google-cloud-aiplatform  \\\n",
         "                                 google-cloud-storage \\\n",
-        "                                 'kfp<2' \\\n",
-        "                                 'google-cloud-pipeline-components<2'"
+        "                                 kfp \\\n",
+        "                                 google-cloud-pipeline-components"
       ]
     },
     {
@@ -272,10 +272,8 @@
       },
       "outputs": [],
       "source": [
-        "IS_COLAB = False\n",
         "# from google.colab import auth\n",
-        "# auth.authenticate_user()\n",
-        "# IS_COLAB = True"
+        "# auth.authenticate_user()"
       ]
     },
     {
@@ -299,6 +297,17 @@
         "Create a storage bucket to store intermediate artifacts such as datasets."
       ]
     },
+    {
+      "cell_type": "code",
+      "execution_count": null,
+      "metadata": {
+        "id": "MzGDU7TWdts_"
+      },
+      "outputs": [],
+      "source": [
+        "BUCKET_URI = f\"gs://your-bucket-name-{PROJECT_ID}-unique\"  # @param {type:\"string\"}"
+      ]
+    },
     {
       "cell_type": "markdown",
       "metadata": {
@@ -312,11 +321,11 @@
       "cell_type": "code",
       "execution_count": null,
       "metadata": {
-        "id": "MzGDU7TWdts_"
+        "id": "NIq7R4HZCfIc"
       },
       "outputs": [],
       "source": [
-        "BUCKET_URI = f\"gs://your-bucket-name-{PROJECT_ID}-unique\"  # @param {type:\"string\"}"
+        "! gsutil mb -l {REGION} -p {PROJECT_ID} {BUCKET_URI}"
       ]
     },
     {
@@ -349,6 +358,9 @@
       },
       "outputs": [],
       "source": [
+        "import sys\n",
+        "\n",
+        "IS_COLAB = \"google.colab\" in sys.modules\n",
         "if (\n",
         "    SERVICE_ACCOUNT == \"\"\n",
         "    or SERVICE_ACCOUNT is None\n",
@@ -414,9 +426,8 @@
         "import json\n",
         "\n",
         "import google.cloud.aiplatform as aip\n",
-        "from kfp import dsl\n",
-        "from kfp.v2 import compiler\n",
-        "from kfp.v2.dsl import component"
+        "from kfp import compiler, dsl\n",
+        "from kfp.dsl import component"
       ]
     },
     {
@@ -571,24 +582,24 @@
         "\n",
         "        args_generator = args_generator_op()\n",
         "        with dsl.ParallelFor(args_generator.output) as item:\n",
-        "            print_op(json_string)\n",
+        "            print_op(msg=json_string)\n",
         "\n",
         "            with dsl.Condition(flip1.output == \"heads\", name=\"heads\"):\n",
-        "                print_op(item.cats)\n",
+        "                print_op(msg=item.cats)\n",
         "\n",
         "            with dsl.Condition(flip1.output == \"tails\", name=\"tails\"):\n",
-        "                print_op(item.dogs)\n",
+        "                print_op(msg=item.dogs)\n",
         "\n",
         "    with dsl.ParallelFor(json_string) as item:\n",
         "        with dsl.Condition(item.snakes == \"boa\", name=\"snakes\"):\n",
-        "            print_op(item.snakes)\n",
-        "            print_op(item.lizards)\n",
-        "            print_op(item.bunnies)\n",
+        "            print_op(msg=item.snakes)\n",
+        "            print_op(msg=item.lizards)\n",
+        "            print_op(msg=item.bunnies)\n",
         "\n",
         "    # it is possible to access sub-items\n",
         "    with dsl.ParallelFor(json_string) as item:\n",
         "        with dsl.ParallelFor(item.bunnies) as item_bunnies:\n",
-        "            print_op(item_bunnies.cottontail)"
+        "            print_op(msg=item_bunnies.cottontail)"
       ]
     },
     {
@@ -610,10 +621,8 @@
       },
       "outputs": [],
       "source": [
-        "from kfp.v2 import compiler  # noqa: F811\n",
-        "\n",
         "compiler.Compiler().compile(\n",
-        "    pipeline_func=pipeline, package_path=\"control_pipeline.json\".replace(\" \", \"_\")\n",
+        "    pipeline_func=pipeline, package_path=\"control_pipeline.yaml\"\n",
         ")"
       ]
     },
@@ -640,7 +649,7 @@
         "\n",
         "job = aip.PipelineJob(\n",
         "    display_name=DISPLAY_NAME,\n",
-        "    template_path=\"control_pipeline.json\".replace(\" \", \"_\"),\n",
+        "    template_path=\"control_pipeline.yaml\",\n",
         "    pipeline_root=PIPELINE_ROOT,\n",
         ")\n",
         "\n",
@@ -679,16 +688,7 @@
         "To clean up all Google Cloud resources used in this project, you can [delete the Google Cloud\n",
         "project](https://cloud.google.com/resource-manager/docs/creating-managing-projects#shutting_down_projects) you used for the tutorial.\n",
         "\n",
-        "Otherwise, you can delete the individual resources you created in this tutorial -- *Note:* this is auto-generated and not all resources may be applicable for this tutorial:\n",
-        "\n",
-        "- Dataset\n",
-        "- Pipeline\n",
-        "- Model\n",
-        "- Endpoint\n",
-        "- Batch Job\n",
-        "- Custom Job\n",
-        "- Hyperparameter Tuning Job\n",
-        "- Cloud Storage Bucket"
+        "Otherwise, you can delete the individual resources you created in this tutorial."
       ]
     },
     {
@@ -699,93 +699,11 @@
       },
       "outputs": [],
       "source": [
-        "delete_dataset = True\n",
-        "delete_pipeline = True\n",
-        "delete_model = True\n",
-        "delete_endpoint = True\n",
-        "delete_batchjob = True\n",
-        "delete_customjob = True\n",
-        "delete_hptjob = True\n",
-        "delete_bucket = True\n",
+        "import os\n",
         "\n",
-        "try:\n",
-        "    if delete_model and \"DISPLAY_NAME\" in globals():\n",
-        "        models = aip.Model.list(\n",
-        "            filter=f\"display_name={DISPLAY_NAME}\", order_by=\"create_time\"\n",
-        "        )\n",
-        "        model = models[0]\n",
-        "        aip.Model.delete(model)\n",
-        "        print(\"Deleted model:\", model)\n",
-        "except Exception as e:\n",
-        "    print(e)\n",
+        "delete_bucket = False\n",
         "\n",
-        "try:\n",
-        "    if delete_endpoint and \"DISPLAY_NAME\" in globals():\n",
-        "        endpoints = aip.Endpoint.list(\n",
-        "            filter=f\"display_name={DISPLAY_NAME}_endpoint\", order_by=\"create_time\"\n",
-        "        )\n",
-        "        endpoint = endpoints[0]\n",
-        "        endpoint.undeploy_all()\n",
-        "        aip.Endpoint.delete(endpoint.resource_name)\n",
-        "        print(\"Deleted endpoint:\", endpoint)\n",
-        "except Exception as e:\n",
-        "    print(e)\n",
-        "\n",
-        "if delete_dataset and \"DISPLAY_NAME\" in globals():\n",
-        "    if \"none\" == \"tabular\":\n",
-        "        try:\n",
-        "            datasets = aip.TabularDataset.list(\n",
-        "                filter=f\"display_name={DISPLAY_NAME}\", order_by=\"create_time\"\n",
-        "            )\n",
-        "            dataset = datasets[0]\n",
-        "            aip.TabularDataset.delete(dataset.resource_name)\n",
-        "            print(\"Deleted dataset:\", dataset)\n",
-        "        except Exception as e:\n",
-        "            print(e)\n",
-        "\n",
-        "    if \"none\" == \"image\":\n",
-        "        try:\n",
-        "            datasets = aip.ImageDataset.list(\n",
-        "                filter=f\"display_name={DISPLAY_NAME}\", order_by=\"create_time\"\n",
-        "            )\n",
-        "            dataset = datasets[0]\n",
-        "            aip.ImageDataset.delete(dataset.resource_name)\n",
-        "            print(\"Deleted dataset:\", dataset)\n",
-        "        except Exception as e:\n",
-        "            print(e)\n",
-        "\n",
-        "    if \"none\" == \"text\":\n",
-        "        try:\n",
-        "            datasets = aip.TextDataset.list(\n",
-        "                filter=f\"display_name={DISPLAY_NAME}\", order_by=\"create_time\"\n",
-        "            )\n",
-        "            dataset = datasets[0]\n",
-        "            aip.TextDataset.delete(dataset.resource_name)\n",
-        "            print(\"Deleted dataset:\", dataset)\n",
-        "        except Exception as e:\n",
-        "            print(e)\n",
-        "\n",
-        "    if \"none\" == \"video\":\n",
-        "        try:\n",
-        "            datasets = aip.VideoDataset.list(\n",
-        "                filter=f\"display_name={DISPLAY_NAME}\", order_by=\"create_time\"\n",
-        "            )\n",
-        "            dataset = datasets[0]\n",
-        "            aip.VideoDataset.delete(dataset.resource_name)\n",
-        "            print(\"Deleted dataset:\", dataset)\n",
-        "        except Exception as e:\n",
-        "            print(e)\n",
-        "\n",
-        "try:\n",
-        "    if delete_pipeline and \"DISPLAY_NAME\" in globals():\n",
-        "        pipelines = aip.PipelineJob.list(\n",
-        "            filter=f\"display_name={DISPLAY_NAME}\", order_by=\"create_time\"\n",
-        "        )\n",
-        "        pipeline = pipelines[0]\n",
-        "        aip.PipelineJob.delete(pipeline.resource_name)\n",
-        "        print(\"Deleted pipeline:\", pipeline)\n",
-        "except Exception as e:\n",
-        "    print(e)\n",
+        "job.delete()\n",
         "\n",
         "if delete_bucket or os.getenv(\"IS_TESTING\"):\n",
         "    ! gsutil rm -r $BUCKET_URI"