mirror of
https://github.com/GoogleCloudPlatform/vertex-ai-samples.git
synced 2026-09-26 14:42:04 +00:00
* chore: refactor according to template, removes boilerplate, adds colab enterprise * refactore: adds testing variables * fix, chore: end to end testing with version change as fix * chore: lint * chore: addresses review comments and runs lint
38 KiB
38 KiB
In [ ]:
# Copyright 2022 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# https://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.In [ ]:
! pip3 install --upgrade --quiet google-cloud-aiplatform \
google-cloud-bigquery \
tensorflow \
tensorflow-io \
xgboost \
numpy \
pandas \
pyarrow \
db-dtypesIn [ ]:
import sys
if "google.colab" in sys.modules:
import IPython
app = IPython.Application.instance()
app.kernel.do_shutdown(True)In [ ]:
import sys
if "google.colab" in sys.modules:
from google.colab import auth
auth.authenticate_user()In [ ]:
PROJECT_ID = "[your-project-id]" # @param {type:"string"}
LOCATION = "us-central1" # @param {type:"string"}
from google.cloud import aiplatform
aiplatform.init(project=PROJECT_ID, location=LOCATION)In [ ]:
BUCKET_URI = f"gs://your-bucket-name-{PROJECT_ID}-unique" # @param {type:"string"}In [ ]:
! gsutil mb -l $LOCATION $BUCKET_URIIn [ ]:
import pandas as pd
import xgboost as xgb
from google.cloud import bigqueryIn [ ]:
bqclient = bigquery.Client(project=PROJECT_ID)In [ ]:
IMPORT_FILE = "bq://bigquery-public-data.samples.gsod"
BQ_TABLE = "bigquery-public-data.samples.gsod"In [ ]:
dataset = aiplatform.TabularDataset.create(
display_name="NOAA historical weather data",
bq_source=[IMPORT_FILE],
labels={"user_metadata": BUCKET_URI[5:]},
)
label_column = "mean_temp"
print(dataset.resource_name)In [ ]:
comps = BQ_TABLE.split(".")
BQ_PROJECT_DATASET_TABLE = comps[0] + ":" + comps[1] + "." + comps[2]
! bq --location=us extract --destination_format CSV $BQ_PROJECT_DATASET_TABLE $BUCKET_URI/mydata*.csv
IMPORT_FILES = ! gsutil ls $BUCKET_URI/mydata*.csv
print(IMPORT_FILES)
EXAMPLE_FILE = IMPORT_FILES[0]
! gsutil cat $EXAMPLE_FILE | headIn [ ]:
gcs_source = IMPORT_FILES
dataset = aiplatform.TabularDataset.create(
display_name="NOAA historical weather data",
gcs_source=gcs_source,
labels={"user_metadata": BUCKET_URI[5:]},
)
label_column = "mean_temp"
print(dataset.resource_name)In [ ]:
# Set dataset name and view name in BigQuery
BQ_MY_DATASET = "[your-dataset-name]"
BQ_MY_TABLE = "[your-view-name]"
# Otherwise, use the default names
if (
BQ_MY_DATASET == ""
or BQ_MY_DATASET is None
or BQ_MY_DATASET == "[your-dataset-name]"
):
BQ_MY_DATASET = "mlops_dataset"
if BQ_MY_TABLE == "" or BQ_MY_TABLE is None or BQ_MY_TABLE == "[your-view-name]":
BQ_MY_TABLE = "mlops_view"In [ ]:
# Create the resources
! bq --location=US mk -d \
$PROJECT_ID:$BQ_MY_DATASET
sql_script = f'''
CREATE OR REPLACE VIEW `{PROJECT_ID}.{BQ_MY_DATASET}.{BQ_MY_TABLE}`
AS SELECT station_number,year,month,day,mean_temp FROM `{BQ_TABLE}`
'''
print(sql_script)
query = bqclient.query(sql_script)In [ ]:
# Download the table.
table = bigquery.TableReference.from_string(BQ_TABLE)
rows = bqclient.list_rows(
table,
max_results=500,
selected_fields=[
bigquery.SchemaField("station_number", "STRING"),
bigquery.SchemaField("year", "INTEGER"),
bigquery.SchemaField("month", "INTEGER"),
bigquery.SchemaField("day", "INTEGER"),
bigquery.SchemaField("mean_temp", "FLOAT"),
],
)
dataframe = rows.to_dataframe()
print(dataframe.head())In [ ]:
from tensorflow.python.framework import dtypes
from tensorflow_io.bigquery import BigQueryClient
feature_names = "station_number,year,month,day".split(",")
target_name = "mean_temp"
def read_bigquery(project, dataset, table):
tensorflow_io_bigquery_client = BigQueryClient()
read_session = tensorflow_io_bigquery_client.read_session(
parent="projects/" + PROJECT_ID,
project_id=project,
dataset_id=dataset,
table_id=table,
selected_fields=feature_names + [target_name],
output_types=[dtypes.string] + [dtypes.int32] * 3 + [dtypes.float32],
requested_streams=2,
)
dataset = read_session.parallel_read_rows()
return dataset
PROJECT, DATASET, TABLE = IMPORT_FILE.split("/")[-1].split(".")
tf_dataset = read_bigquery(PROJECT, DATASET, TABLE)
print(tf_dataset.take(1))In [ ]:
import tensorflow as tf
feature_names = ["station_number,year,month,day".split(",")]
target_name = "mean_temp"
tf_dataset = tf.data.experimental.CsvDataset(
filenames=IMPORT_FILES,
header=True,
select_cols=feature_names.append(target_name),
record_defaults=[dtypes.string] + [dtypes.int32] * 3 + [dtypes.float32],
)
print(tf_dataset.take(1))In [ ]:
LOCATION = "us"
SCHEMA = [
bigquery.SchemaField("station_number", "STRING"),
bigquery.SchemaField("year", "INTEGER"),
bigquery.SchemaField("month", "INTEGER"),
bigquery.SchemaField("day", "INTEGER"),
bigquery.SchemaField("mean_temp", "FLOAT"),
]
DATASET_ID = "samples"
TABLE_ID = "gsod"
def create_bigquery_dataset(dataset_id):
dataset = bigquery.Dataset(
bigquery.dataset.DatasetReference(PROJECT_ID, dataset_id)
)
dataset.location = "us"
try:
dataset = bqclient.create_dataset(dataset) # API request
return True
except Exception as err:
print(err)
if err.code != 409: # http_client.CONFLICT
raise
return False
def load_data_into_bigquery(dataframe, dataset_id, table_id):
create_bigquery_dataset(dataset_id)
dataset = bqclient.dataset(dataset_id)
table = dataset.table(table_id)
job_config = bigquery.LoadJobConfig(
# Specify a (partial) schema. All columns are always written to the
# table. The schema is used to assist in data type definitions.
schema=[
bigquery.SchemaField("station_number", "STRING"),
bigquery.SchemaField("year", "INTEGER"),
bigquery.SchemaField("month", "INTEGER"),
bigquery.SchemaField("day", "INTEGER"),
bigquery.SchemaField("mean_temp", "FLOAT"),
],
# Optionally, set the write disposition. BigQuery appends loaded rows
# to an existing table by default, but with WRITE_TRUNCATE write
# disposition it replaces the table with the loaded data.
write_disposition="WRITE_TRUNCATE",
)
NEW_BQ_TABLE = f"{PROJECT_ID}.{dataset_id}.{table_id}"
job = bqclient.load_table_from_dataframe(
dataframe, NEW_BQ_TABLE, job_config=job_config
) # Make an API request.
job.result() # Wait for the job to complete.
table = bqclient.get_table(NEW_BQ_TABLE) # Make an API request.
print(
"Loaded {} rows and {} columns to {}".format(
table.num_rows, len(table.schema), NEW_BQ_TABLE
)
)
load_data_into_bigquery(dataframe, DATASET_ID, TABLE_ID)In [ ]:
LOCATION = "us"
CSV_SCHEMA = [
bigquery.SchemaField("station_number", "STRING"),
bigquery.SchemaField("wban_number", "STRING"),
bigquery.SchemaField("year", "INTEGER"),
bigquery.SchemaField("month", "INTEGER"),
bigquery.SchemaField("day", "INTEGER"),
bigquery.SchemaField("mean_temp", "FLOAT"),
bigquery.SchemaField("num_mean_temp_samples", "INTEGER"),
bigquery.SchemaField("mean_dew_point", "FLOAT"),
bigquery.SchemaField("num_mean_dew_point_samples", "INTEGER"),
bigquery.SchemaField("mean_sealevel_pressure", "FLOAT"),
bigquery.SchemaField("num_mean_sealevel_pressure_samples", "INTEGER"),
bigquery.SchemaField("mean_station_pressure", "FLOAT"),
bigquery.SchemaField("num_mean_station_pressure_samples", "INTEGER"),
bigquery.SchemaField("mean_visibility", "FLOAT"),
bigquery.SchemaField("num_mean_visibility_samples", "INTEGER"),
bigquery.SchemaField("mean_wind_speed", "FLOAT"),
bigquery.SchemaField("num_mean_wind_speed_samples", "INTEGER"),
bigquery.SchemaField("max_sustained_wind_speed", "FLOAT"),
bigquery.SchemaField("max_gust_wind_speed", "FLOAT"),
bigquery.SchemaField("max_temperature", "FLOAT"),
bigquery.SchemaField("max_temperature_explicit", "BOOLEAN"),
bigquery.SchemaField("min_temperature", "FLOAT"),
bigquery.SchemaField("min_temperature_explicit", "BOOLEAN"),
bigquery.SchemaField("total_percipitation", "FLOAT"),
bigquery.SchemaField("snow_depth", "FLOAT"),
bigquery.SchemaField("fog", "BOOLEAN"),
bigquery.SchemaField("rain", "BOOLEAN"),
bigquery.SchemaField("snow", "BOOLEAN"),
bigquery.SchemaField("hail", "BOOLEAN"),
bigquery.SchemaField("thunder", "BOOLEAN"),
bigquery.SchemaField("tornado", "BOOLEAN"),
]
DATASET_ID = "samples"
TABLE_ID = "gsod"
def load_data_into_bigquery(url, dataset_id, table_id):
create_bigquery_dataset(dataset_id)
dataset = bqclient.dataset(dataset_id)
table = dataset.table(table_id)
job_config = bigquery.LoadJobConfig()
job_config.write_disposition = bigquery.WriteDisposition.WRITE_TRUNCATE
job_config.source_format = bigquery.SourceFormat.CSV
job_config.schema = CSV_SCHEMA
job_config.skip_leading_rows = 1 # heading
load_job = bqclient.load_table_from_uri(url, table, job_config=job_config)
print("Starting job {}".format(load_job.job_id))
load_job.result() # Waits for table load to complete.
print("Job finished.")
destination_table = bqclient.get_table(table)
print("Loaded {} rows.".format(destination_table.num_rows))
load_data_into_bigquery(IMPORT_FILES, DATASET_ID, TABLE_ID)In [ ]:
dataframe["station_number"] = pd.to_numeric(dataframe["station_number"])
labels = dataframe["mean_temp"]
data = dataframe.drop(["mean_temp"], axis=1)
dtrain = xgb.DMatrix(data, label=labels)In [ ]:
! gsutil cp $EXAMPLE_FILE data.csv
dtrain = xgb.DMatrix("data.csv?format=csv&label_column=4")In [ ]:
import os
# Delete the dataset using the Vertex dataset object
dataset.delete()
# Delete the temporary BigQuery dataset
! bq rm -r -f $PROJECT_ID:$DATASET_ID
delete_storage = False
if delete_storage or os.getenv("IS_TESTING"):
# Delete the created GCS bucket
! gsutil rm -r $BUCKET_URI
# Delete the created BigQuery datasets
! bq rm -r -f $PROJECT_ID:$BQ_MY_DATASET

