feat: Job and BatchPredictionJob classes

[vinnysenthil]

PR for batch prediction method in Model class to follow once this PR and #66 get merged.

Includes the following:

• Job base class for remaining Job subclasses
  • Adds a shared status() method for Job
• BatchPredictionJob class
  • Adds a iter_outputs() method which returns either a BQ QueryJob or list of GCS Blobs
• 7 unit tests covering 94% of jobs.py
• Addition of dependency google-cloud-bigquery
• Addition of a constants.py to have single source of truth on re-used constants (i.e. supported parameters, API constants)

Fixes b/169783178, b/171074104 🦕

[sasha-gitg]

LGTM! Minor requests.

[vinnysenthil]

Hey @tswast , looping you in here for your BigQuery expertise.

Context:
This is for the Model Builder SDK and this PR implements a BatchPredictionJob class where users can send many instances to a trained model and get many predictions back. There is an option provide a BQ project to write prediction output directly into. The service creates a new dataset in the provided BQ project and writes the predictions into a table.

We're implementing a helper method iter_outputs for users to get an iterable object to traverse the predictions. In the scenario they chose a BQ output, I return a QueryJob that runs SELECT * {generated_dataset_name}.predictions and allows a user to pass in an int value for LIMIT.

Here are the lines in question:

• https://github.com/googleapis/python-aiplatform/pull/79/files#diff-cc0613c901a9b3ee26167824eeaf859c223b19f1c4c0c48dc01423fb9815e7c5R187-R206
• https://github.com/googleapis/python-aiplatform/pull/79/files#diff-5469ef5507566e6890ac7691b9379a5ce66c72e21b7f657484905fedd076e3b4R234-R243
• https://github.com/googleapis/python-aiplatform/pull/79/files#diff-5469ef5507566e6890ac7691b9379a5ce66c72e21b7f657484905fedd076e3b4R176-R185

PTAL and share your thoughts when you get a chance 😄

[tswast]

LGTM for BigQuery usage. I have some concerns about constructing both Storage and BigQuery clients within a client method, but depending on the performance expectations of this could be considered a minor issue.

[tswast]

I'm a bit concerned about this. Ideally you'd re-use the credentials from uCAIP client.

Also, there's a risk of leaking sockets when you create clients on-the-fly. Not as big a deal for REST clients, but definitely a concern for gRPC clients. googleapis/google-cloud-python#9790 googleapis/google-cloud-python#9457

[vinnysenthil]

Suggested change

	bq_client = bigquery.Client()
	bq_client = bigquery.Client(
	credentials=self.api_client._transport._credentials
	)

^ This change would build a BigQuery Client using the same credentials as uCAIP's JobServiceClient.

In regards to the leaking sockets, would the solution referenced in that issue work? See below

# Close sockets opened by BQ Client
bq_client._http._auth_request.session.close()
bq_client._http.close()

[tswast]

This change for credentials LGTM. (Storage should get similar treatment).

It's a little trickier in our case, because we want the client to live for the lifetime of the RowIterator.

Unless you want to convert to full list of rows / pandas dataframe before returning? In which case all the API requests would be made here and we could close the client when done (FWIW, the client does have a close function in BQ. https://googleapis.dev/python/bigquery/latest/generated/google.cloud.bigquery.client.Client.html#google.cloud.bigquery.client.Client.close)

[vinnysenthil]

Made the credentials change on both BQ and Storage.

Re: closing connections - this is indeed tricky since the method is meant to return an iterator. However your comment made realize a larger issue of us instantiating a GAPIC client for every instance of a high-level SDK object. I'm capturing this in b/174111905.

Will merge this blocking PR for now, thanks for calling this issue out!