Getting Started with Image Classification Using Vertex AI and BigQuery

This guide provides a complete end-to-end workflow for training models and classifying imagery assets using Google Cloud's Vertex AI platform with Gemini 2.5 Flash. You'll learn to integrate BigQuery for data retrieval, Cloud Storage for asset management, and Vertex AI for machine learning inference in a Python Colab environment.

Configuration

Set the following project-specific variables before running the code samples:

PROJECT_ID = "PROJECT_ID"
REGION = "REGION "  # e.g., "us-central1"
LOCATION = "LOCATION "  # e.g., "us"
CUSTOMER_ID = "CUSTOMER_ID" # required to subscribe to the dataset

Environment Setup

Install required dependencies and configure authentication to access Google Cloud services:

# Install Google Cloud SDK dependencies for AI Platform integration
!pip install google-cloud-aiplatform google-cloud-storage google-cloud-bigquery google-cloud-bigquery-data-exchange -q

# Import core libraries for cloud services and machine learning operations
import json
import os
from google.cloud import bigquery
import vertexai
from vertexai.generative_models import GenerativeModel, Part

# Configure authentication for Google Cloud service access
# Initiates OAuth flow in new browser tab if authentication required
from google.colab import auth

if os.environ.get("VERTEX_PRODUCT") != "COLAB_ENTERPRISE":
  from google.colab import auth
  auth.authenticate_user(project_id=PROJECT_ID)

# Initialize Vertex AI client with project configuration
vertexai.init(project=PROJECT_ID, location=REGION)

print(f"Vertex AI initialized for project: {PROJECT_ID} in region: {REGION}")

Subscribe to the Analytics Hub dataset

You must also subscribe to the Analytics Hub dataset.

from google.cloud import bigquery_data_exchange_v1beta1

ah_client = bigquery_data_exchange_v1beta1.AnalyticsHubServiceClient()

HUB_PROJECT_ID = 'maps-platform-analytics-hub'
DATA_EXCHANGE_ID = f"imagery_insights_exchange_{LOCATION}"
LINKED_DATASET_NAME = f"imagery_insights___preview___{LOCATION}"


# subscribe to the listing (create a linked dataset in your consumer project)
destination_dataset = bigquery_data_exchange_v1beta1.DestinationDataset()
destination_dataset.dataset_reference.dataset_id = LINKED_DATASET_NAME
destination_dataset.dataset_reference.project_id = PROJECT_ID
destination_dataset.location = LOCATION
LISTING_ID=f"imagery_insights_{CUSTOMER_ID.replace('-', '_')}__{LOCATION}"

published_listing = f"projects/{HUB_PROJECT_ID}/locations/{LOCATION}/dataExchanges/{DATA_EXCHANGE_ID}/listings/{LISTING_ID}"

request = bigquery_data_exchange_v1beta1.SubscribeListingRequest(
    destination_dataset=destination_dataset,
    name=published_listing,
)

# request the subscription
ah_client.subscribe_listing(request=request)

Data Extraction with BigQuery

Execute a BigQuery query to extract Google Cloud Storage URIs from the latest_observations table. These URIs will be passed directly to the Vertex AI model for classification.

# Initialize BigQuery client
bigquery_client = bigquery.Client(project=PROJECT_ID)

# Define SQL query to retrieve observation records from imagery dataset
query = f"""
SELECT
 *
FROM
 `{PROJECT_ID}.imagery_insights___preview___{LOCATION}.latest_observations`
LIMIT 10;
"""

print(f"Executing BigQuery query:\n{query}")

# Submit query job to BigQuery service and await completion
query_job = bigquery_client.query(query)

# Transform query results into structured data format for downstream processing
# Convert BigQuery Row objects to dictionary representations for enhanced accessibility
query_response_data = []
for row in query_job:
   query_response_data.append(dict(row))

# Extract Cloud Storage URIs from result set, filtering null values
gcs_uris = [item.get("gcs_uri") for item in query_response_data if item.get("gcs_uri")]

print(f"BigQuery query returned {len(query_response_data)} records.")
print(f"Extracted {len(gcs_uris)} GCS URIs:")
for uri in gcs_uris:
   print(uri)

Image Classification Function

This helper function handles the classification of images using Vertex AI's Gemini 2.5 Flash model:

def classify_image_with_gemini(gcs_uri: str, prompt: str = "What is in this image?") -> str:
   """
   Performs multimodal image classification using Vertex AI's Gemini 2.5 Flash model.

   Leverages direct Cloud Storage integration to process image assets without local
   download requirements, enabling scalable batch processing workflows.

   Args:
       gcs_uri (str): Fully qualified Google Cloud Storage URI 
                     (format: gs://bucket-name/path/to/image.jpg)
       prompt (str): Natural language instruction for classification task execution

   Returns:
       str: Generated textual description from the generative model, or error message
            if classification pipeline fails

   Raises:
       Exception: Captures service-level errors and returns structured failure response
   """
   try:
       # Instantiate Gemini 2.5 Flash model for inference operations
       model = GenerativeModel("gemini-2.5-flash")

       # Construct multimodal Part object from Cloud Storage reference
       # Note: MIME type may need dynamic inference for mixed image formats
       image_part = Part.from_uri(uri=gcs_uri, mime_type="image/jpeg")

       # Execute multimodal inference request with combined visual and textual inputs
       responses = model.generate_content([image_part, prompt])
       return responses.text
   except Exception as e:
       print(f"Error classifying image from URI {gcs_uri}: {e}")
       return "Classification failed."

Batch Image Classification

Process all extracted URIs and generate classifications:

classification_results = []

# Execute batch classification pipeline across all extracted GCS URIs
for uri in gcs_uris:
   print(f"\nProcessing: {uri}")

   # Define comprehensive classification prompt for detailed feature extraction
   classification_prompt = "Describe this image in detail, focusing on any objects, signs, or features visible."

   # Invoke Gemini model for multimodal inference on current asset
   result = classify_image_with_gemini(uri, classification_prompt)

   # Aggregate structured results for downstream analytics and reporting
   classification_results.append({"gcs_uri": uri, "classification": result})

   print(f"Classification for {uri}:\n{result}")

Next Steps

With your images classified, consider these advanced workflows:

  • Model Fine-tuning: Use classification results to train custom models.
  • Automated Processing: Set up Cloud Functions to classify new images automatically.
  • Data Analysis: Perform statistical analysis on classification patterns.
  • Integration: Connect results to downstream applications.

Troubleshooting

Common issues and solutions:

  • Authentication errors: Ensure proper IAM roles and API enablement.
  • Rate limiting: Implement exponential backoff for large batches.
  • Memory constraints: Process images in smaller batches for large datasets.
  • URI format errors: Verify GCS URIs follow the format gs://bucket-name/path/to/image.

For additional support, refer to the Vertex AI documentation and BigQuery documentation.