Merge branch 'main' into batch_tabular_model

notebooks/community/ml_ops/stage6/get_started_with_custom_tabular_model_batch.ipynb

@@ -716,14 +716,19 @@
         "\n",
         "### Input format for batch prediction jobs\n",
         "\n",
-        "The batch server accepts the following input formats:\n",
+        "The batch server accepts the following input formats for custom tabular models:\n",
         "\n",
         "- JSONL\n",
         "- CSV\n",
-        "- TFRecords\n",
-        "- File-List\n",
         "- BigQuery table\n",
         "\n",
+        "### Output format for batch prediction jobs\n",
+        "\n",
+        "The batch server accepts the following output formats for custom tabular models:\n",
+        "\n",
+        "- JSONL\n",
+        "- BigQuery table (when input is BigQuery table)\n",
+        "\n",
         "### Pivot format\n",
         "\n",
         "The batch server converts the input format to the `pivot` (JSONL) format as follows:\n",
@@ -1098,12 +1103,12 @@
       "cell_type": "code",
       "execution_count": null,
       "metadata": {
-        "id": "0d078e5e8953"
+        "id": "776346c90074"
       },
       "outputs": [],
       "source": [
-        "DATA_DIR = \"gs://cloud-samples-data/ai-platform/iris/iris_data.csv\"\n",
-        "! gsutil cat $data_dir | head -n 10 > test.csv\n",
+        "data_file = \"gs://cloud-samples-data/ai-platform/iris/iris_data.csv\"\n",
+        "! gsutil cat $data_file | head -n 10 > test.csv\n",
         "\n",
         "! cat test.csv\n",
         "\n",
@@ -1209,6 +1214,264 @@
         "            break"
       ]
     },
+    {
+      "cell_type": "markdown",
+      "metadata": {
+        "id": "9ce65fff362b"
+      },
+      "source": [
+        "#### Delete the batch prediction job\n",
+        "\n",
+        "You can delete your batch prediction job using the `delete()` method."
+      ]
+    },
+    {
+      "cell_type": "code",
+      "execution_count": null,
+      "metadata": {
+        "id": "5943dfde5123"
+      },
+      "outputs": [],
+      "source": [
+        "batch_prediction_job.delete()"
+      ]
+    },
+    {
+      "cell_type": "markdown",
+      "metadata": {
+        "id": "d61289b87070"
+      },
+      "source": [
+        "## Batch prediction with BigQuery input format\n",
+        "\n",
+        "Next, you do the same batch job, except the input format is a BigQuery table. When the input format is a BigQuery table, the output format has to be a BigQuery table as well. To use BigQuery as the input format, your model must take its input as a list (array) of values. "
+      ]
+    },
+    {
+      "cell_type": "markdown",
+      "metadata": {
+        "id": "csv_to_bq"
+      },
+      "source": [
+        "### Create a BigQuery dataset from CSV files\n",
+        "\n",
+        "You can create a BigQuery dataset from CSV files using the BigQuery `create_dataset()` and `load_table_from_uri()` methods, as follows:\n",
+        "\n",
+        "- `create_dataset()`: Creates an empty BigQuery dataset, with the following parameters:\n",
+        " - `dataset_ref`: The `DatasetReference` created from the dataset_id -- e.g., samples.\n",
+        "- `load_table_from_uri()`: Loads one or more CSV files into a table within the corresponding dataset, with the following parameters:\n",
+        " - `url`: A set of one or more CVS files in Cloud Storage storage.\n",
+        " - `table`: The `TableReference` for the table.\n",
+        " - `job_config`: Specifications on how to load the CSV data.\n",
+        "\n",
+        "Learn more about [Importing CSV data into BigQuery](https://www.tensorflow.org/io/tutorials/bigquery#import_census_data_into_bigquery)."
+      ]
+    },
+    {
+      "cell_type": "code",
+      "execution_count": null,
+      "metadata": {
+        "id": "csv_to_bq"
+      },
+      "outputs": [],
+      "source": [
+        "LOCATION = \"us\"\n",
+        "\n",
+        "CSV_SCHEMA = [\n",
+        "    bigquery.SchemaField(\"sepal_length\", \"FLOAT\"),\n",
+        "    bigquery.SchemaField(\"sepal_width\", \"FLOAT\"),\n",
+        "    bigquery.SchemaField(\"petal_length\", \"FLOAT\"),\n",
+        "    bigquery.SchemaField(\"petal_width\", \"FLOAT\"),\n",
+        "]\n",
+        "\n",
+        "DATASET_ID = \"batch\"\n",
+        "TABLE_ID = \"slice1\"\n",
+        "\n",
+        "\n",
+        "def create_bigquery_dataset(dataset_id):\n",
+        "    dataset = bigquery.Dataset(\n",
+        "        bigquery.dataset.DatasetReference(PROJECT_ID, dataset_id)\n",
+        "    )\n",
+        "    dataset.location = \"us\"\n",
+        "\n",
+        "    try:\n",
+        "        dataset = bqclient.create_dataset(dataset)  # API request\n",
+        "        return True\n",
+        "    except Exception as err:\n",
+        "        print(err)\n",
+        "        if err.code != 409:  # http_client.CONFLICT\n",
+        "            raise\n",
+        "    return False\n",
+        "\n",
+        "\n",
+        "def load_data_into_bigquery(url, dataset_id, table_id):\n",
+        "    create_bigquery_dataset(dataset_id)\n",
+        "    dataset = bqclient.dataset(dataset_id)\n",
+        "    table = dataset.table(table_id)\n",
+        "\n",
+        "    job_config = bigquery.LoadJobConfig()\n",
+        "    job_config.write_disposition = bigquery.WriteDisposition.WRITE_TRUNCATE\n",
+        "    job_config.source_format = bigquery.SourceFormat.CSV\n",
+        "    job_config.schema = CSV_SCHEMA\n",
+        "    job_config.skip_leading_rows = 1  # heading\n",
+        "\n",
+        "    load_job = bqclient.load_table_from_uri(url, table, job_config=job_config)\n",
+        "    print(\"Starting job {}\".format(load_job.job_id))\n",
+        "\n",
+        "    load_job.result()  # Waits for table load to complete.\n",
+        "    print(\"Job finished.\")\n",
+        "\n",
+        "    destination_table = bqclient.get_table(table)\n",
+        "    print(\"Loaded {} rows.\".format(destination_table.num_rows))\n",
+        "\n",
+        "    return destination_table\n",
+        "\n",
+        "\n",
+        "bq_table = load_data_into_bigquery(data_file, DATASET_ID, TABLE_ID)\n",
+        "print(bq_table)"
+      ]
+    },
+    {
+      "cell_type": "markdown",
+      "metadata": {
+        "id": "send_prediction_request:image"
+      },
+      "source": [
+        "### Send the batch prediction request\n",
+        "\n",
+        "Again, you make a batch prediction request with the `batch_predict()` method, but with the following changes in parameters:\n",
+        "\n",
+        "- `instances_format`: Set to 'bigquery'\n",
+        "- `predictions_format`: Set to 'bigquery'\n",
+        "- `bigquery_source`: Used instead of `gcs_source`.\n",
+        "- `bigquery_destination_prefix`: Used instead of `gcs_destination_prefix`"
+      ]
+    },
+    {
+      "cell_type": "code",
+      "execution_count": null,
+      "metadata": {
+        "id": "1cf1076178fc"
+      },
+      "outputs": [],
+      "source": [
+        "MIN_NODES = 1\n",
+        "MAX_NODES = 1\n",
+        "\n",
+        "# The name of the job\n",
+        "BATCH_PREDICTION_JOB_NAME = \"churn_batch-\" + UUID\n",
+        "\n",
+        "# Folder in the bucket to write results to\n",
+        "DESTINATION_FOLDER = \"batch_prediction_results_bq\"\n",
+        "\n",
+        "# The Cloud Storage bucket to upload results to\n",
+        "BATCH_PREDICTION_GCS_DEST_PREFIX = BUCKET_URI + \"/\" + DESTINATION_FOLDER\n",
+        "\n",
+        "BQ_TABLE_SOURCE = f\"bq://{PROJECT_ID}.{bq_table.dataset_id}.{bq_table.table_id}\"\n",
+        "\n",
+        "# Make SDK batch_predict method call\n",
+        "batch_prediction_job = model.batch_predict(\n",
+        "    instances_format=\"bigquery\",\n",
+        "    predictions_format=\"bigquery\",\n",
+        "    job_display_name=BATCH_PREDICTION_JOB_NAME,\n",
+        "    bigquery_source=BQ_TABLE_SOURCE,\n",
+        "    bigquery_destination_prefix=f\"bq://{PROJECT_ID}.batch\",\n",
+        "    model_parameters=None,\n",
+        "    machine_type=DEPLOY_COMPUTE,\n",
+        "    accelerator_type=DEPLOY_GPU,\n",
+        "    accelerator_count=DEPLOY_NGPU,\n",
+        "    starting_replica_count=MIN_NODES,\n",
+        "    max_replica_count=MAX_NODES,\n",
+        "    sync=True,\n",
+        ")"
+      ]
+    },
+    {
+      "cell_type": "markdown",
+      "metadata": {
+        "id": "get_batch_prediction:mbsdk,custom,icn"
+      },
+      "source": [
+        "### Get the predictions\n",
+        "\n",
+        "Next, get the results from the completed batch prediction job.\n",
+        "\n",
+        "The results are written to a BigQuery table at the BigQuery dataset path you specified as the destination. The batch server creates the table, where the table location is specified by:\n",
+        "\n",
+        "`batch_prediction_job.output_info.bigquery_output_dataset`: The project and dataset components.\n",
+        "`batch_prediction_job.output_info.bigquery_output_table`: The table component.\n"
+      ]
+    },
+    {
+      "cell_type": "code",
+      "execution_count": null,
+      "metadata": {
+        "id": "cd515ed616c8"
+      },
+      "outputs": [],
+      "source": [
+        "BQ_RESULTS_TABLE = (\n",
+        "    batch_prediction_job.output_info.bigquery_output_dataset\n",
+        "    + \".\"\n",
+        "    + batch_prediction_job.output_info.bigquery_output_table\n",
+        ")\n",
+        "\n",
+        "print(BQ_RESULTS_TABLE)\n",
+        "\n",
+        "table = bigquery.TableReference.from_string(BQ_RESULTS_TABLE[5:])\n",
+        "\n",
+        "rows = bqclient.list_rows(table, max_results=10)\n",
+        "\n",
+        "for row in rows:\n",
+        "    print(row)\n",
+        "    for key, value in row.items():\n",
+        "        pass"
+      ]
+    },
+    {
+      "cell_type": "markdown",
+      "metadata": {
+        "id": "c7cda79d0e42"
+      },
+      "source": [
+        "#### Delete the batch prediction job\n",
+        "\n",
+        "You can delete your batch prediction job using the `delete()` method."
+      ]
+    },
+    {
+      "cell_type": "code",
+      "execution_count": null,
+      "metadata": {
+        "id": "de4b5c638c2a"
+      },
+      "outputs": [],
+      "source": [
+        "batch_prediction_job.delete()"
+      ]
+    },
+    {
+      "cell_type": "markdown",
+      "metadata": {
+        "id": "e5d4b9c51e86"
+      },
+      "source": [
+        "#### Delete the model\n",
+        "\n",
+        "You can delete your model using the `delete()` method."
+      ]
+    },
+    {
+      "cell_type": "code",
+      "execution_count": null,
+      "metadata": {
+        "id": "00fa6a7b4f24"
+      },
+      "outputs": [],
+      "source": [
+        "model.delete()"
+      ]
+    },
     {
       "cell_type": "markdown",
       "metadata": {
@@ -1232,16 +1495,6 @@
       "outputs": [],
       "source": [
         "delete_bucket = False\n",
-        "delete_model = True\n",
-        "delete_batch_job = True\n",
-        "\n",
-        "if delete_model:\n",
-        "    try:\n",
-        "        model.delete()\n",
-        "    except Exception as e:\n",
-        "        print(e)\n",
-        "if delete_batch_job:\n",
-        "    batch_prediction_job.delete()\n",
         "\n",
         "if delete_bucket or os.getenv(\"IS_TESTING\"):\n",
         "    ! gsutil rm -rf {BUCKET_URI}"