feat: stage 3 notebooks

notebooks/community/ml_ops/stage1/mlops_data_management.ipynb

@@ -82,16 +82,16 @@
         "\n",
         "This tutorial uses the following Vertex AI and Data Analytics services:\n",
         "\n",
-        "- `Vertex Datasets`\n",
+        "- `Vertex AI Datasets`\n",
         "- `BigQuery`\n",
         "- `Dataflow`\n",
         "\n",
         "The steps performed include:\n",
         "\n",
         "- Explore and visualize the data.\n",
-        "- Create a Vertex `Dataset` resource from `BigQuery` table -- for AutoML training.\n",
+        "- Create a Vertex AI `Dataset` resource from `BigQuery` table -- for AutoML training.\n",
         "- Extract a copy of the dataset to a CSV file in Cloud Storage.\n",
-        "- Create a Vertex `Dataset` resource from CSV files -- alternative for AutoML training.\n",
+        "- Create a Vertex AI `Dataset` resource from CSV files -- alternative for AutoML training.\n",
         "- Read a sample of the `BigQuery` dataset into a dataframe.\n",
         "- Generate statistics and data schema using TensorFlow Data Validation from the samples in the dataframe.\n",
         "- Generate a TFRecord feature specification using TensorFlow Data Validation from the data schema.\n",
@@ -110,7 +110,7 @@
         "\n",
         " - For large amounts of data, use BigQuery table. Otherwise, use a CSV file stored in Cloud Storage.\n",
         " - When storing a large amount of data in CSV file, shard the data at 10,000 rows per shard.\n",
-        " - Create a managed dataset with Vertex `TabularDataset`.\n",
+        " - Create a managed dataset with Vertex AI `TabularDataset`.\n",
         " - Preprocess the data with `Dataflow`"
       ]
     },
@@ -144,7 +144,8 @@
         "    ! pip3 install --upgrade google-cloud-logging $USER_FLAG\n",
         "    ! pip3 install --upgrade apache-beam[gcp] $USER_FLAG\n",
         "    ! pip3 install --upgrade pyarrow $USER_FLAG\n",
-        "    ! pip3 install --upgrade cloudml-hypertune $USER_FLAG"
+        "    ! pip3 install --upgrade cloudml-hypertune $USER_FLAG\n",
+        "    ! pip3 install --upgrade kfp $USER_FLAG"
       ]
     },
     {
@@ -501,9 +502,9 @@
         "id": "init_aip:mbsdk"
       },
       "source": [
-        "### Initialize Vertex SDK for Python\n",
+        "### Initialize Vertex AI SDK for Python\n",
         "\n",
-        "Initialize the Vertex SDK for Python for your project and corresponding bucket."
+        "Initialize the Vertex AI SDK for Python for your project and corresponding bucket."
       ]
     },
     {
@@ -1223,10 +1224,12 @@
         "    return len(ratio) - 1\n",
         "\n",
         "\n",
-        "def convert_to_jsonl(data):\n",
+        "def convert_to_jsonl(data, label=None):\n",
         "    \"\"\"Converts a parsed record to JSON\"\"\"\n",
         "    import json\n",
         "\n",
+        "    if label:\n",
+        "        del data[label]\n",
         "    return json.dumps(data)\n",
         "\n",
         "\n",
@@ -1236,6 +1239,7 @@
         "    pipeline_options = beam.pipeline.PipelineOptions(flags=[], **args)\n",
         "\n",
         "    raw_data_query = args[\"raw_data_query\"]\n",
+        "    label = args[\"label\"]\n",
         "    transformed_data_prefix = args[\"transformed_data_prefix\"]\n",
         "    transform_artifact_dir = args[\"transform_artifact_dir\"]\n",
         "    exported_jsonl_prefix = args[\"exported_jsonl_prefix\"]\n",
@@ -1345,16 +1349,6 @@
         "                transform_artifact_dir\n",
         "            )\n",
         "\n",
-        "            # Convert raw test data to JSON (for batch prediction)\n",
-        "            json_test_data = (raw_test_data) | \"Convert Test Data\" >> beam.Map(\n",
-        "                convert_to_jsonl\n",
-        "            )\n",
-        "\n",
-        "            # Write raw test data to GCS as JSONL files.\n",
-        "            _ = json_test_data | \"Write JSONL Test Data\" >> beam.io.WriteToText(\n",
-        "                file_path_prefix=exported_jsonl_prefix, file_name_suffix=\".jsonl\"\n",
-        "            )\n",
-        "\n",
         "            # Write raw test data to GCS as TF Records\n",
         "            _ = (\n",
         "                raw_test_data\n",
@@ -1366,6 +1360,16 @@
         "                )\n",
         "            )\n",
         "\n",
+        "            # Convert raw test data to JSON (for batch prediction)\n",
+        "            json_test_data = (raw_test_data) | \"Convert Batch Test Data\" >> beam.Map(\n",
+        "                convert_to_jsonl, label=label\n",
+        "            )\n",
+        "\n",
+        "            # Write raw test data to GCS as JSONL files.\n",
+        "            _ = json_test_data | \"Write JSONL Test Data\" >> beam.io.WriteToText(\n",
+        "                file_path_prefix=exported_jsonl_prefix, file_name_suffix=\".jsonl\"\n",
+        "            )\n",
+        "\n",
         "\n",
         "EXPORTED_JSONL_PREFIX = os.path.join(BUCKET_NAME, \"exported_data/jsonl\")\n",
         "EXPORTED_TFREC_PREFIX = os.path.join(BUCKET_NAME, \"exported_data/tfrec\")\n",
@@ -1378,6 +1382,7 @@
         "args = {\n",
         "    \"runner\": RUNNER,\n",
         "    \"raw_data_query\": QUERY_STRING,\n",
+        "    \"label\": label_column,\n",
         "    \"transformed_data_prefix\": TRANSFORMED_DATA_PREFIX,\n",
         "    \"transform_artifact_dir\": TRANSFORM_ARTIFACTS_DIR,\n",
         "    \"exported_jsonl_prefix\": EXPORTED_JSONL_PREFIX,\n",
@@ -1396,7 +1401,7 @@
         "! gsutil ls $TRANSFORMED_DATA_PREFIX/val\n",
         "! gsutil ls $TRANSFORMED_DATA_PREFIX/test\n",
         "! gsutil ls $TRANSFORM_ARTIFACTS_DIR\n",
-        "! gsutil ls $EXPORTED_JSONL_PREFIX\n",
+        "! gsutil ls {EXPORTED_JSONL_PREFIX}*\n",
         "! gsutil ls $EXPORTED_TFREC_PREFIX"
       ]
     },