Build a Workspace Usage Data Pipeline

Pull Google Workspace metrics into BigQuery with a scheduled Cloud Run job.

Introduction

30 Second Summary

Every organization running Google Workspace generates usage data across Gmail, Drive, Meet, and more. Without a way to collect and analyze that data, IT leaders and consultants have no visibility into adoption trends or license utilization.

In this project, you will build a production-ready data pipeline that pulls Google Workspace usage metrics from the Admin SDK Reports API and loads them into BigQuery on a daily schedule.

What You'll Build

A scheduled Cloud Run job that authenticates via domain-wide delegation, pulls Customer Usage reports from Google's Admin SDK Reports API, flattens them into a clean schema, and loads them into a date-partitioned BigQuery table using idempotent writes.

By the end of this project, you'll have:

  • A Python pipeline that authenticates with a service account, pulls usage data from the Google Admin SDK Reports API, and loads it into BigQuery.
  • A date-partitioned BigQuery table with a flexible metric-per-row schema designed for aggregate usage metrics.
  • A Cloud Run job deployed from source and triggered daily by Cloud Scheduler, with lag-aware date windowing built in.
  • Secret Mission: Make the pipeline config-driven so onboarding a new client requires zero code changes.

Are there any prerequisites?

You need a GCP project with billing enabled, the gcloud CLI installed, and Python 3.10+ installed.

You also need access to a Google Workspace organization where an admin can grant domain-wide delegation to your service account.

Set Up GCP Resources and Service Account

Your pipeline needs three things before it can pull any data: a service account that authenticates with Google's APIs, permission for that account to read Workspace usage data, and a BigQuery dataset to land the results in.

In this step, you will create all three resources inside your GCP project. By the end, your pipeline's foundation is fully in place.

In this step, get ready to:
  • Create a service account, download its JSON key, and copy its Client ID.
  • Enable the Admin SDK API and send delegation instructions to your Workspace admin.
  • Create a BigQuery dataset to store your pipeline output.
Create a service account and download the JSON key

A service account is a non-human identity that your pipeline uses to authenticate with Google APIs. You will create one, generate a key file for local authentication, and record the Client ID needed for domain-wide delegation.

  • Open the Google Cloud Console in your browser.
  • Make sure your project is selected in the project dropdown at the top of the page.
  • Click the hamburger menu in the top-left corner.
  • Navigate to IAM & Admin, then select Service Accounts.
  • Click Create Service Account at the top of the page.
  • In the Service account name field, enter workspace-usage-pipeline.
  • Leave the optional role and access fields blank for now. Click Done to finish creating the account.

You should see workspace-usage-pipeline listed on the Service Accounts page.

Next, generate a JSON key file so your local script can authenticate as this service account.

  • Click the email address of the workspace-usage-pipeline service account to open its details.
  • Click the Keys tab.
  • Click Add Key, then select Create new key.
  • Select JSON as the key type and click Create.
  • A JSON file downloads automatically. Rename it to workspace-usage-pipeline-key.json and move it into your project's working directory.

Why download a JSON key?

The JSON key file contains the private key your script uses to prove it is the service account. This file is the only copy. Store it securely and never commit it to version control.

Now copy the Client ID, which your Workspace admin will need for delegation.

  • Go back to the service account's details page by clicking the Details tab.
  • Click Advanced settings.
  • Under the Domain-wide delegation section, copy the Client ID value. Paste it somewhere safe (a text file or note) for the next substep.

Finally, enable the Admin SDK API so your service account can call the Reports endpoints.

  • In the left sidebar, navigate to APIs & Services, then click Enable APIs and services.
  • Search for Admin SDK API in the search bar.
  • Click the Admin SDK API result, then click Enable.

What is the Admin SDK API?

The Admin SDK Reports API is the endpoint your pipeline calls to retrieve Workspace usage metrics. Enabling it tells GCP that your project is allowed to make requests to this API.

Don't see the Enable button?

If the button says Manage instead, the API is already enabled on your project. You can continue to the next substep.

If you see a permissions error, confirm you are signed in as the project owner or have the serviceusage.services.enable permission on this project.

Why can't I enable the Admin SDK API on my project?

Send delegation instructions to your Workspace admin

Domain-wide delegation allows your service account to impersonate a Workspace admin user when calling the Reports API. Your organization's Workspace admin needs to authorize this in the Google Admin Console.

Send your Workspace admin the following two pieces of information:

  • The Client ID you copied in the previous substep.
  • The single OAuth scope: https://www.googleapis.com/auth/admin.reports.usage.readonly.

Why only one scope?

The principle of least privilege applies here. By granting only the admin.reports.usage.readonly scope, the service account can read usage reports but cannot modify any Workspace settings, read emails, or access user data beyond aggregate metrics.

Share the following instructions with your admin (or follow them yourself if you are the Workspace admin):

  • Open the Google Admin Console at admin.google.com.
  • Navigate to Security, then Access and data control, then API controls.
  • Click Manage Domain Wide Delegation.
  • Click Add new.
  • Paste the Client ID into the Client ID field.
  • In the OAuth scopes field, paste https://www.googleapis.com/auth/admin.reports.usage.readonly.
  • Click Authorize.

Delegation usually activates within minutes

Google states propagation can take up to 24 hours, but in practice delegation is available within a few minutes. If your admin has completed the steps, you can continue with the project and verify access in Step 2.

Create a BigQuery dataset for pipeline output

Your pipeline needs a destination in BigQuery to store the usage metrics it pulls. A dataset is the top-level container that holds your tables.

  • Create the workspace_usage dataset in the US region by running this command:
bq mk --dataset --location=US [[PROJECT_ID="your-gcp-project-id"]]:workspace_usage

What does this command do?

The bq mk --dataset command creates a new BigQuery dataset. The --location=US flag pins the dataset to the US multi-region, and the PROJECT_ID:workspace_usage format ensures the dataset is created in your specific project.

  • Verify the dataset was created by running:
bq ls [[PROJECT_ID="your-gcp-project-id"]]:

You should see workspace_usage listed in the output.

Dataset not showing up?

Make sure you replaced the project ID placeholder with your actual GCP project ID. You can find it at the top of the Google Cloud Console.

If you see a permission error, confirm that the BigQuery API is enabled on your project and that your gcloud CLI is authenticated with an account that has BigQuery Data Editor permissions.

Why can't I create a BigQuery dataset in my project?

Your GCP foundation is ready. You have a service account with a key file, delegation granted for usage reports, and a BigQuery dataset waiting for data. Next up, you will prove that the authentication works by pulling live data from the Reports API.

Prove Access Locally and Inspect Data Shape

You have a service account, domain-wide delegation, and a BigQuery dataset waiting for data. Before building the full pipeline, you need to prove that the authentication chain works and see what the raw API response actually looks like.

This step is about reducing risk. If you discover a problem (wrong scope, delegation not propagated, unexpected data shape), you want to find it now in a quick script, not after deploying infrastructure.

In this step, get ready to:
  • Set up a Python virtual environment and install the Google client libraries.
  • Write and run an exploration script that pulls Customer Usage and User Usage data from the Admin SDK Reports API.
  • Inspect the JSON response structure and understand Google's data lag.
Check your Python version

This project requires Python 3.10 or higher because the google-auth library depends on it.

  • Check your Python version by running this command:
python3 --version

You should see output in the format Python 3.x.y (for example, Python 3.12.4).

✔️ I see version 3.10 or higher

You are ready to go. Continue to the next substep.

ⓧ I see a version lower than 3.10

Your system Python is too old for this project. On macOS, the default python3 may point to an older version even if you have a newer one installed.

brew update
brew install pyenv
  • Install Python 3.10 and set it for your project by running:
pyenv install 3.10
pyenv local 3.10
  • Run python3 --version again to confirm you now see 3.10 or higher.

On Windows, download Python 3.10+ from python.org and run the installer.

ⓧ Command not found

Python 3 is not installed on your system.

  • On macOS, install Python via Homebrew by running:
brew install python@3.10
  • On Windows, download Python 3.10+ from python.org and run the installer.
  • On Linux, use your package manager (e.g. sudo apt install python3.10) or install pyenv from https://github.com/pyenv/pyenv.
  • After installation, run python3 --version again to confirm.
Set up your project directory and Python environment
  • Move to your Desktop by running this command:
cd ~/Desktop
  • Create your project folder and move into it by running these commands:
mkdir workspace-usage-pipeline
cd workspace-usage-pipeline
  • Copy your service account key file into this folder. The file should be named workspace-usage-pipeline-key.json (the one you downloaded in Step 1).
  • Copy it by running this command (adjust the source path if you saved it somewhere else):
cp [[KEY_SOURCE_PATH="path/to/your/downloaded/workspace-usage-pipeline-key.json"]] .
  • Confirm the file is in your project folder by running:
ls

You should see workspace-usage-pipeline-key.json listed in the output.

  • Create and activate a Python virtual environment by running:
python3 -m venv .venv
source .venv/bin/activate

What is a virtual environment?

A virtual environment is an isolated Python installation for your project. Packages you install here won't conflict with other Python projects on your system.

You'll know it's active when your terminal prompt shows (.venv) at the beginning.

On Windows?

Replace source .venv/bin/activate with .venv\Scripts\activate.

help me activate a Python virtual environment on Windows

  • Create a requirements.txt file and install the dependencies by running these commands:
cat > requirements.txt << EOF
google-api-python-client
google-auth
google-cloud-bigquery
EOF
pip install -r requirements.txt

What are these packages?

  • google-api-python-client provides the Python client for Google APIs, including the Admin SDK Reports API.
  • google-auth handles service account authentication and credential delegation.
  • google-cloud-bigquery is the client library for loading data into BigQuery.

You should see pip download and install several packages. Look for Successfully installed at the end of the output.

Seeing permission errors or install failures?

Make sure your virtual environment is active (you should see (.venv) in your prompt). If not, re-run source .venv/bin/activate.

help me fix pip install errors in my virtual environment

Write and run an exploration script to pull Customer Usage data

Now you'll write a script that authenticates as your service account, impersonates a Workspace admin, and pulls one day of usage data from the Admin SDK Reports API.

  • Create a file called explore.py in your project folder and paste the following code:
import json
from google.oauth2 import service_account
from googleapiclient.discovery import build

# Path to your service account key file
SERVICE_ACCOUNT_FILE = 'workspace-usage-pipeline-key.json'

# The scope required for usage reports
SCOPES = ['https://www.googleapis.com/auth/admin.reports.usage.readonly']

# Email of a Workspace admin to impersonate
ADMIN_EMAIL = '[[ADMIN_EMAIL="your-workspace-admin@yourdomain.com"]]'

# Authenticate with service account and delegate to admin
credentials = service_account.Credentials.from_service_account_file(
    SERVICE_ACCOUNT_FILE, scopes=SCOPES
)
delegated_credentials = credentials.with_subject(ADMIN_EMAIL)

# Build the Admin SDK Reports service
service = build('admin', 'reports_v1', credentials=delegated_credentials)

# Pull Customer Usage report for a test date (at least 5 days ago)
print('=== Customer Usage Report ===')
customer_response = service.customerUsageReports().get(date='2026-06-05').execute()
print(json.dumps(customer_response, indent=2))

What does this code do?

  • from_service_account_file() loads your JSON key and creates credentials scoped to the read-only usage scope.
  • with_subject() tells Google to impersonate the admin email you provide. This is how domain-wide delegation works for service accounts.
  • build('admin', 'reports_v1', ...) creates a client for the Admin SDK Reports API.
  • customerUsageReports().get(date=...) fetches aggregate usage metrics for your entire Workspace organization on that date.

Choose a date at least 5 days in the past

Google has a 3-5 day data lag for usage reports. If you use a recent date (like yesterday), you'll get empty or partial data. The example uses 2026-06-05. Replace it with a date at least 5 days before today in YYYY-MM-DD format.

  • Save explore.py and run the script:
python3 explore.py

You should see a JSON response printed to your terminal containing a usageReports array with parameter objects.

Seeing a 403 or authorization error?

This usually means domain-wide delegation hasn't propagated yet, or the scope was entered incorrectly.

  • Double-check that your admin granted the exact scope https://www.googleapis.com/auth/admin.reports.usage.readonly (no extra spaces).
  • Confirm the ADMIN_EMAIL in your script is a super-admin in your Workspace org.
  • Delegation can take up to 24 hours to propagate, though it usually takes minutes.

help me debug a 403 error when calling the Google Admin SDK Reports API with domain-wide delegation

Add User Usage data and inspect the response structure

The Customer Usage report gives org-level aggregates. The User Usage report gives per-user metrics, including fields like userEmail that you must NOT store in your pipeline (to protect PII).

  • Open explore.py and add the following code at the bottom, after the Customer Usage section:
# Pull User Usage report for the same date
print('\n=== User Usage Report ===')
user_response = service.userUsageReport().get(userKey='all', date='2026-06-05').execute()
print(json.dumps(user_response, indent=2))

What does this call do?

userUsageReport().get(userKey='all', ...) fetches usage data for every user in your org for that date. The response includes per-user fields like entity.userEmail that reveal PII.

In the production pipeline, you will aggregate these metrics (counts and sums) before loading to BigQuery, ensuring no PII is stored.

  • Save explore.py and run it again:
python3 explore.py

This time you'll see both the Customer Usage and User Usage responses.

  • Look at the JSON structure. Each report entry in the usageReports array contains:
  • entity object with customerId, type, and (for User Usage) userEmail.
  • parameters array where each entry has a name and a value field (intValue, stringValue, or msgValue).
  • nextPageToken field if the response spans multiple pages (you'll handle pagination in the full pipeline).

✔️ Awesome, I've got everything!

Great. Make sure you've saved explore.py with both the Customer Usage and User Usage calls.

ⓧ I'd like to double check the full code

Here is the complete explore.py file:

import json
from google.oauth2 import service_account
from googleapiclient.discovery import build

# Path to your service account key file
SERVICE_ACCOUNT_FILE = 'workspace-usage-pipeline-key.json'

# The scope required for usage reports
SCOPES = ['https://www.googleapis.com/auth/admin.reports.usage.readonly']

# Email of a Workspace admin to impersonate
ADMIN_EMAIL = '[[ADMIN_EMAIL="your-workspace-admin@yourdomain.com"]]'

# Authenticate with service account and delegate to admin
credentials = service_account.Credentials.from_service_account_file(
    SERVICE_ACCOUNT_FILE, scopes=SCOPES
)
delegated_credentials = credentials.with_subject(ADMIN_EMAIL)

# Build the Admin SDK Reports service
service = build('admin', 'reports_v1', credentials=delegated_credentials)

# Pull Customer Usage report for a test date (at least 5 days ago)
print('=== Customer Usage Report ===')
customer_response = service.customerUsageReports().get(date='2026-06-05').execute()
print(json.dumps(customer_response, indent=2))

# Pull User Usage report for the same date
print('\n=== User Usage Report ===')
user_response = service.userUsageReport().get(userKey='all', date='2026-06-05').execute()
print(json.dumps(user_response, indent=2))
Understand why local proof matters

Before building deployment infrastructure, you now have concrete answers to several critical questions.

  • Auth works: your service account, delegation, and scope are correctly configured.
  • Data shape is known: you've seen the nested parameters structure and know you need to flatten it into rows for BigQuery.
  • PII boundaries are clear: User Usage includes userEmail which you will aggregate away before loading.
  • Data lag is real: dates within the last 3-5 days return empty or partial data. Your pipeline must account for this.

Why does Google have a 3-5 day data lag?

Usage metrics are based on UTC-8:00 (Pacific Standard Time) and are compiled in batch. Google needs time to aggregate activity across all services and users. The API retains the last 450 days of data, so there's no rush to pull recent dates.

In the pipeline you'll build next, the default date window will end 3 days ago to avoid requesting data that isn't available yet.

Your authentication is proven and you understand the data shape. Next up, you'll design a BigQuery schema that can hold these metrics in a flexible, query-friendly format.

Design BigQuery Schema and Create Table

You have a working connection to the Admin SDK Reports API and you know what the data looks like. Now you need a place to land that data in BigQuery.

The schema you choose determines how flexible your pipeline is over time. A poor schema choice means you need to run DDL changes every time Google adds a new metric. In this step, you will design a schema that handles any metric without modification, then create the table.

In this step, get ready to:
  • Design a long/narrow metric-per-row schema for aggregate usage data.
  • Create a date-partitioned BigQuery table using a Python script.
Design the metric-per-row schema

When choosing a schema for metrics data, you have two broad options: a wide table (one column per metric) or a long/narrow table (one row per metric).

A wide table would need a new column added every time Google introduces a metric. The long/narrow approach stores each metric as its own row, meaning new metrics land automatically with zero schema changes.

Your table will use the following fields:

  • report_date (DATE) the date the usage data is for. This is also the partition key.
  • client_id (STRING) identifies the Workspace customer.
  • app_name (STRING) the Google Workspace application, e.g. gmail, drive, meet.
  • metric_name (STRING) the specific metric, e.g. num_1day_active_users.
  • metric_value (FLOAT64) the numeric value of the metric.
  • loaded_at (TIMESTAMP) records when the row was inserted into BigQuery.

The table uses date-based partitioning on the report_date column with DAY granularity. This means BigQuery physically separates data by date, making queries for specific date ranges fast and cheap.

Why a long/narrow schema instead of a wide table?

The Admin SDK Reports API returns dozens of metrics per application, and Google adds new ones periodically. A wide table (one column per metric) would require a DDL change every time a new metric appears.

With a metric-per-row schema, adding a new metric means zero code or schema changes. The pipeline flattens whatever parameters the API returns into rows that fit the existing columns.

Create the table with a Python script

You will use your AI coding assistant to generate a Python script that creates the usage_metrics table in your workspace_usage dataset. The script uses the google-cloud-bigquery library you already installed.

  • Ask your AI assistant to generate a Python script that creates a BigQuery table with the schema above and DAY-granularity time partitioning on report_date. Use a prompt like:
Write a Python script called create_table.py that:
- Uses google-cloud-bigquery to create a table named usage_metrics in the workspace_usage dataset
- Uses my gcloud project (get it from the client)
- Has these fields: report_date (DATE, REQUIRED), client_id (STRING, REQUIRED), app_name (STRING, REQUIRED), metric_name (STRING, REQUIRED), metric_value (FLOAT64, NULLABLE), loaded_at (TIMESTAMP, REQUIRED)
- Partitions by report_date with DAY granularity
- Prints a confirmation message with the table name and partition field

Your AI assistant should generate something similar to the following. Verify that your output matches this structure before saving:

from google.cloud import bigquery

# Initialize BigQuery client (uses your gcloud project)
client = bigquery.Client()

# Define the full table reference: project.dataset.table
table_id = f"{client.project}.workspace_usage.usage_metrics"

# Define the schema for aggregate usage metrics
schema = [
    bigquery.SchemaField("report_date", "DATE", mode="REQUIRED"),
    bigquery.SchemaField("client_id", "STRING", mode="REQUIRED"),
    bigquery.SchemaField("app_name", "STRING", mode="REQUIRED"),
    bigquery.SchemaField("metric_name", "STRING", mode="REQUIRED"),
    bigquery.SchemaField("metric_value", "FLOAT64", mode="NULLABLE"),
    bigquery.SchemaField("loaded_at", "TIMESTAMP", mode="REQUIRED"),
]

# Create the table with date-based partitioning on report_date
table = bigquery.Table(table_id, schema=schema)
table.time_partitioning = bigquery.TimePartitioning(
    type_=bigquery.TimePartitioningType.DAY,
    field="report_date",
)

# Send the create request to BigQuery
table = client.create_table(table)
print(f"Created table {table.project}.{table.dataset_id}.{table.table_id}")
print(f"Partitioned on: {table.time_partitioning.field}")

What does this script do?

  • bigquery.Client() picks up your active GCP project from gcloud config automatically.
  • The table_id uses the fully-qualified format project.dataset.table that the BigQuery Python client requires.
  • TimePartitioning with type_=bigquery.TimePartitioningType.DAY tells BigQuery to create a separate physical partition for each date value in report_date.
  • client.create_table(table) sends the create request to BigQuery and returns the created table object.
  • Save the file as create_table.py in your project directory.
  • Run the script by executing this command:
python3 create_table.py
  • You should see output confirming the table was created, including the table name and the partition field report_date.
  • Verify the table schema and partitioning config by running:
bq show workspace_usage.usage_metrics

You should see the six columns listed with their types, and a Time Partitioning field showing DAY-level partitioning on report_date.

Table already exists error?

If you see an error saying the table already exists, it means you ran the script more than once. You can delete the table with bq rm workspace_usage.usage_metrics and run the script again.

If bq show shows an unexpected schema, double-check that the field names and types in your script match the schema listed above exactly.

help me debug my BigQuery table creation script

✔️ Awesome, I've got everything!

Great. Double check you have saved create_table.py and confirmed the table exists with bq show.

ⓧ I'd like to double check the full code

Here is the complete create_table.py file:

from google.cloud import bigquery

# Initialize BigQuery client (uses your gcloud project)
client = bigquery.Client()

# Define the full table reference: project.dataset.table
table_id = f"{client.project}.workspace_usage.usage_metrics"

# Define the schema for aggregate usage metrics
schema = [
    bigquery.SchemaField("report_date", "DATE", mode="REQUIRED"),
    bigquery.SchemaField("client_id", "STRING", mode="REQUIRED"),
    bigquery.SchemaField("app_name", "STRING", mode="REQUIRED"),
    bigquery.SchemaField("metric_name", "STRING", mode="REQUIRED"),
    bigquery.SchemaField("metric_value", "FLOAT64", mode="NULLABLE"),
    bigquery.SchemaField("loaded_at", "TIMESTAMP", mode="REQUIRED"),
]

# Create the table with date-based partitioning on report_date
table = bigquery.Table(table_id, schema=schema)
table.time_partitioning = bigquery.TimePartitioning(
    type_=bigquery.TimePartitioningType.DAY,
    field="report_date",
)

# Send the create request to BigQuery
table = client.create_table(table)
print(f"Created table {table.project}.{table.dataset_id}.{table.table_id}")
print(f"Partitioned on: {table.time_partitioning.field}")

Your table is ready to receive data. Next up, you will build the main pipeline script that pulls data from the Reports API and loads it into this table using idempotent writes.

Build the Pipeline Script

You have a BigQuery table ready to receive data and you have confirmed authentication works. Now it is time to build the actual pipeline script that ties everything together.

The script needs to pull usage data from the Admin SDK Reports API, transform the nested response into flat rows, and load them into your partitioned table. It also needs to be smart about Google's 3-5 day data lag so it never fails on unavailable data.

In this step, get ready to:
  • Build a pipeline script that pulls, flattens, and loads usage data into BigQuery.
  • Add error handling and lag-aware date windowing so the pipeline runs reliably.
  • Test locally to confirm data lands in the correct BigQuery partitions.
Build the main pipeline script with your AI assistant

The pipeline script handles four responsibilities: reading a date range, pulling API data with pagination, flattening the nested response, and loading rows into BigQuery partitions idempotently.

  • Create a new file called main.py in your project folder (the same folder that contains explore.py and requirements.txt).
  • Open your AI coding assistant (Cursor, Claude, or ChatGPT) and give it the following prompt. Paste it as-is, replacing the placeholder values with your actual project ID and admin email:
Write a Python script called main.py that:

1. Reads START_DATE and END_DATE from environment variables. If not set, defaults to today minus 5 days through today minus 3 days (to account for Google's 3-5 day data lag).

2. Authenticates using a service account JSON key file (workspace-usage-pipeline-key.json) with scope https://www.googleapis.com/auth/admin.reports.usage.readonly and impersonates a Workspace admin email via .with_subject().

3. Builds the Admin SDK Reports service using build('admin', 'reports_v1', credentials=delegated_credentials).

4. For each date in the range, calls service.customerUsageReports().get(date=current_date).execute() and handles pagination by following nextPageToken.

5. Flattens each usage report's parameters into rows with these fields:
   - report_date (the date string)
   - client_id (from entity.customerId)
   - app_name (parsed from parameter name before the colon, e.g. 'gmail' from 'gmail:num_1day_active_users')
   - metric_name (the full parameter name)
   - metric_value (from intValue, cast to float; skip parameters without intValue)
   - loaded_at (current UTC timestamp)

6. Loads rows into BigQuery using the partition decorator pattern: table_id$YYYYMMDD (e.g. workspace_usage.usage_metrics$20260605). Uses LoadJobConfig with write_disposition=WRITE_TRUNCATE so re-runs replace data for that partition.

7. Logs which date it is processing and how many rows were loaded.

8. Handles HTTP 400 errors (data not ready) gracefully by logging a warning and skipping that date.

9. Exits with sys.exit(1) if ALL dates in the range failed (no data loaded for any date).

Project ID: YOUR_PROJECT_ID
Admin email for impersonation: YOUR_ADMIN_EMAIL
BigQuery table: workspace_usage.usage_metrics

What does this prompt ask for?

The prompt covers every architectural decision your pipeline needs. Here is what each piece accomplishes:

  • Lag-aware defaults (today minus 5 through today minus 3) mean the script never requests dates that Google has not yet populated.
  • Pagination via nextPageToken ensures you collect all data even when responses span multiple pages.
  • The partition decorator pattern (table_id$YYYYMMDD) targets a specific day's partition so WRITE_TRUNCATE only erases that one day, not the whole table.
  • Idempotent writes mean you can re-run the pipeline for the same date without duplicating data.
  • Review the AI-generated code. Before saving, verify the critical patterns described below are present.
Verify the key patterns in the generated code

Your AI assistant will produce the full script. You need to confirm the following patterns exist correctly before running it.

  • Check that the authentication block uses service_account.Credentials.from_service_account_file() with scopes and .with_subject() for impersonation:
# Authenticate with service account and impersonate admin
credentials = service_account.Credentials.from_service_account_file(
    SERVICE_ACCOUNT_FILE, scopes=SCOPES
)
delegated_credentials = credentials.with_subject(ADMIN_EMAIL)
  • Check that the service is built with the correct service name and version:
# Build Admin SDK Reports service
service = build('admin', 'reports_v1', credentials=delegated_credentials)
  • Check that pagination follows nextPageToken until exhausted:
# Paginate through all usage report pages
page_token = None
while True:
    response = service.customerUsageReports().get(
        date=current_date, pageToken=page_token
    ).execute()
    # Process response['usageReports']
    page_token = response.get('nextPageToken')
    if not page_token:
        break
  • Check that BigQuery loading uses the partition decorator and WRITE_TRUNCATE:
# Load into specific partition using decorator
partition_date = current_date.replace('-', '')
table_ref = f"{PROJECT_ID}.workspace_usage.usage_metrics${partition_date}"

job_config = bigquery.LoadJobConfig(
    write_disposition=bigquery.WriteDisposition.WRITE_TRUNCATE,
    source_format=bigquery.SourceFormat.NEWLINE_DELIMITED_JSON,
)

Why WRITE_TRUNCATE on a partition?

The WRITE_TRUNCATE disposition erases all existing data before writing the new data. Combined with the partition decorator ($YYYYMMDD), it only erases data for that specific day's partition.

This makes the pipeline idempotent. Running it twice for the same date produces the same result as running it once. No duplicate rows, no stale data left behind.

  • Check that the default date window accounts for lag. It should calculate dates relative to today:
# Default to lag-aware window: today-5 through today-3
today = datetime.date.today()
start_date = os.environ.get('START_DATE', (today - datetime.timedelta(days=5)).isoformat())
end_date = os.environ.get('END_DATE', (today - datetime.timedelta(days=3)).isoformat())
  • Once you have confirmed all patterns are present, save the file as main.py.

AI output missing a pattern?

If your AI assistant omitted pagination or used WRITE_APPEND instead of WRITE_TRUNCATE, paste the specific pattern from above back into the AI conversation and ask it to incorporate it.

Help me fix my pipeline script

Add error handling and test locally

A production pipeline needs to handle missing data gracefully. Google's Reports API returns HTTP 400 when data for a requested date is not yet available. Your script should skip that date with a warning instead of crashing.

  • Verify your main.py includes a try/except around the API call that catches HttpError with status 400 and logs a warning instead of raising:
from googleapiclient.errors import HttpError

# Inside the date loop
try:
    response = service.customerUsageReports().get(
        date=current_date, pageToken=page_token
    ).execute()
except HttpError as e:
    if e.resp.status == 400:
        print(f"WARNING: Data not ready for {current_date}, skipping.")
        break
    raise

Why handle HTTP 400 specifically?

The Reports API returns a 400 status when metrics for a requested date have not been computed yet. This is expected behavior for recent dates (within the last 3-5 days), not a bug in your code.

By catching this specific error and continuing to the next date, the pipeline stays resilient. Other errors (403 auth failures, 500 server errors) still raise so you can investigate them.

  • Verify the script tracks whether any date succeeded, and exits with a non-zero code if all dates failed:
# After processing all dates
if not any_success:
    print("ERROR: All dates failed. No data was loaded.")
    sys.exit(1)

print(f"Pipeline complete. Loaded data for {success_count} date(s).")
sys.exit(0)

Why exit with a non-zero code?

When this script runs as a Cloud Run job, the platform uses the exit code to determine success or failure. A non-zero exit tells Cloud Run the execution failed, which shows up in logs and can trigger alerts.

  • Save main.py and run the pipeline locally by executing this command (make sure your virtual environment is active):
python main.py

You should see log output showing which dates are being processed, how many rows were loaded per date, and possibly a warning for dates within the last 3 days that were skipped.

  • Verify data landed in BigQuery by running this query:
bq query --use_legacy_sql=false 'SELECT report_date, COUNT(*) as row_count FROM workspace_usage.usage_metrics GROUP BY report_date ORDER BY report_date'

You should see rows grouped by date, with dates 5 or more days ago showing data and recent dates (within 3 days) absent or skipped.

Seeing an authentication error?

If you see a 403 or "Not Authorized" error, confirm that domain-wide delegation has propagated. It can take up to 24 hours, but usually resolves within minutes. Also verify the admin email in your script matches the Workspace admin who authorized the delegation.

If you see "Data not ready" warnings for ALL dates including those 5+ days ago, double-check that your Workspace organization has active users generating usage data.

Help me troubleshoot my pipeline

  • Run the pipeline a second time without changing anything. The same data loads again with no duplicates because WRITE_TRUNCATE replaces the partition contents on each run.

✔️ Awesome, I've got everything!

Great. Double check you have saved main.py and confirmed rows are in your BigQuery table.

ⓧ I'd like to double check the full code

Your main.py should contain all of the patterns shown above, combined into a single working script. The exact implementation will vary based on your AI assistant's output, but the key patterns (auth with .with_subject(), build('admin', 'reports_v1', ...), nextPageToken pagination, partition decorator $YYYYMMDD, WRITE_TRUNCATE, HTTP 400 handling, and non-zero exit on total failure) must all be present.

If your script is missing any of these, paste the relevant pattern back into your AI assistant and ask it to incorporate it.

Your pipeline script is working locally and loading data into the correct BigQuery partitions. Next up, you will containerize it and deploy it as a scheduled Cloud Run job so it runs automatically every day.

Deploy as a Scheduled Cloud Run Job

Your pipeline script works locally. Now it needs to run on a schedule without your laptop being open.

In this step, you will containerize the pipeline, deploy it as a Cloud Run job, and attach a Cloud Scheduler trigger so it runs daily. You will also backfill historical data to prove the pipeline handles date ranges end-to-end.

In this step, get ready to:
  • Create a Dockerfile and deploy the pipeline as a Cloud Run job.
  • Schedule the job to run daily with Cloud Scheduler.
  • Run a historical backfill covering 30+ days of usage data.
Create a Dockerfile and deploy to Cloud Run

Cloud Run jobs run containers. You need a Dockerfile that packages your Python script and its dependencies into an image.

The --source . flag sends your source code to Cloud Build, which builds the image remotely. You do not need Docker Desktop installed locally.

  • Create a file called Dockerfile in your project folder (the same directory as main.py) with the following content:
# Use a lightweight Python base image
FROM python:3.10-slim

# Set the working directory inside the container
WORKDIR /app

# Copy all project files into the container
COPY . .

# Install dependencies
RUN pip install --no-cache-dir -r requirements.txt

# Run the pipeline script when the container starts
ENTRYPOINT ["python", "main.py"]

What does this Dockerfile do?

  • FROM python:3.10-slim starts from a minimal Python 3.10 image, keeping the container small.
  • COPY . . brings your main.py, requirements.txt, and service account key into the container.
  • ENTRYPOINT ["python", "main.py"] tells Cloud Run to execute your pipeline script when the job runs.
  • Save Dockerfile.
  • Confirm the file exists by running:
ls

You should see Dockerfile listed alongside main.py and requirements.txt.

Don't see the Dockerfile?

Make sure you created it in the same directory as main.py. The file must be named exactly Dockerfile with a capital D and no file extension.

Help me troubleshoot my missing Dockerfile

  • Deploy the Cloud Run job by running this command:
gcloud run jobs deploy workspace-usage-pipeline \
  --source . \
  --region us-central1 \
  --set-env-vars START_DATE=2026-06-05,END_DATE=2026-06-07 \
  --service-account=workspace-usage-pipeline@[[PROJECT_ID="your-gcp-project-id"]].iam.gserviceaccount.com

What does this command do?

  • --source . uploads your local directory to Cloud Build, which builds the Docker image remotely.
  • --region us-central1 deploys the job in the US Central region.
  • --set-env-vars passes test dates (June 5-7) so you can verify the pipeline works in the cloud.
  • --service-account attaches your service account so the job has permission to call the Admin SDK and write to BigQuery.

The first deploy may take 2-3 minutes while Cloud Build compiles your image.

Why deploy from source?

The --source . pattern lets Cloud Build handle the container image creation on Google's servers. You write a Dockerfile for understanding, but never need to install or run Docker locally.

  • The CLI may prompt you to enable additional APIs (Artifact Registry, Cloud Build). Enter y to confirm each.
  • Once the deploy finishes, test the job by executing it:
gcloud run jobs execute workspace-usage-pipeline --region us-central1 --wait

The --wait flag blocks until the job completes. You should see a success message confirming the execution finished.

  • Verify the data landed in BigQuery by running:
bq query --use_legacy_sql=false \
  'SELECT DISTINCT report_date FROM workspace_usage.usage_metrics ORDER BY 1'

You should see dates from June 5 through June 7 in the output.

Job execution failed?

  • Check the logs with gcloud run jobs executions list --job workspace-usage-pipeline --region us-central1 to see the execution status.
  • If you see a permissions error, confirm that domain-wide delegation is still active for your service account.
  • If dates return empty, remember that data within the last 3-5 days may be unavailable. Try dates further in the past.

Help me troubleshoot my failed Cloud Run job

Create a Cloud Scheduler trigger for daily runs

The job works when you trigger it manually. Now you need it to run automatically every day.

Cloud Scheduler sends an HTTP POST to the Cloud Run API at a cron interval you define. When no START_DATE or END_DATE env vars are set, your script falls back to its lag-aware default window (today minus 5 through today minus 3).

  • Create the scheduler job by running this command:
gcloud scheduler jobs create http workspace-usage-daily \
  --location us-central1 \
  --schedule="0 8 * * *" \
  --uri="https://run.googleapis.com/v2/projects/[[PROJECT_ID="your-gcp-project-id"]]/locations/us-central1/jobs/workspace-usage-pipeline:run" \
  --http-method POST \
  --oauth-service-account-email [[PROJECT_NUMBER="your-gcp-project-number"]]-compute@developer.gserviceaccount.com

What does this command do?

  • --schedule="0 8 * * *" runs the job every day at 8:00 AM UTC.
  • --uri points to the Cloud Run v2 API endpoint that triggers your job.
  • --oauth-service-account-email authenticates the request using your project's default compute service account. Cloud Run requires an OAuth token because the API is hosted on *.googleapis.com.

You can find your project number in the Google Cloud Console dashboard, or by running:

gcloud projects describe [[PROJECT_ID="your-gcp-project-id"]] --format="value(projectNumber)"
  • Force-run the scheduler to confirm it can trigger the Cloud Run job:
gcloud scheduler jobs run workspace-usage-daily --location us-central1
  • Confirm the trigger succeeded by checking the job status:
gcloud scheduler jobs describe workspace-usage-daily --location us-central1

You should see the last run status showing success.

Scheduler trigger failed?

If you see a permission denied error, the compute service account needs the Cloud Run Invoker role on the job. Grant it by running:

gcloud run jobs add-iam-policy-binding workspace-usage-pipeline \
  --region us-central1 \
  --member="serviceAccount:[[PROJECT_NUMBER="your-gcp-project-number"]]-compute@developer.gserviceaccount.com" \
  --role="roles/run.invoker"

If the scheduler job was not found, confirm you created it in us-central1 (the location must match).

Help me troubleshoot my Cloud Scheduler trigger

Run a historical backfill

Daily scheduled runs only capture recent data going forward. To populate BigQuery with historical usage metrics, you can re-execute the job with overridden environment variables covering a wider date range.

Why is this safe to re-run?

Your pipeline uses WRITE_TRUNCATE at the partition level. If data already exists for a date, it gets replaced atomically. Re-runs and backfills never double-count rows.

  • Execute the backfill by running the job with a 30+ day date range:
gcloud run jobs execute workspace-usage-pipeline \
  --region us-central1 \
  --update-env-vars START_DATE=2026-05-01,END_DATE=2026-06-05 \
  --wait

This execution may take a minute or two as it processes each date in the range.

  • Verify the full date range landed in BigQuery:
bq query --use_legacy_sql=false \
  'SELECT DISTINCT report_date FROM workspace_usage.usage_metrics ORDER BY 1'

You should see dates spanning from early May through early June. Some dates within the most recent 3-5 days may be missing due to Google's data lag.

Seeing gaps in your date range?

  • Gaps in the most recent 3-5 days are expected. Google's Reports API has a data processing lag.
  • If entire weeks are missing, check your job logs for HTTP 400 errors. The pipeline gracefully skips those dates, but they indicate data was not available.

Help me understand my backfill date gaps

Your pipeline is now fully deployed and automated. Cloud Scheduler triggers it daily, and you have proven it handles both single-day runs and multi-week backfills. Next up, the Secret Mission challenges you to make this pipeline config-driven for multi-tenant use.

Secret mission

Make It Config-Driven for Multi-Tenant

Your pipeline works for one Workspace organization. But what if you're a consultant or MSP managing ten clients? In this secret mission, you'll restructure the pipeline to read tenant config from a YAML file, so onboarding a new client means adding a config entry and a service account key. Zero code changes.

Clean Up Your Resources

Clean Up Your Resources

Decide whether to keep your resources running, pause them to come back later, or delete them entirely. This project operates within GCP free-tier limits for typical single-tenant daily usage, so ongoing costs are $0/month if left running.

Resources you used:

  • Cloud Scheduler job (workspace-usage-daily in us-central1).
  • Cloud Run job (workspace-usage-pipeline in us-central1).
  • BigQuery dataset and table (workspace_usage.usage_metrics).
  • Service account (workspace-usage-pipeline) and its JSON key file.
  • Local project files (Python scripts, virtual environment, Dockerfile, clients.yaml).

Keep everything running

No action needed. Choose this if you are still actively building or want the pipeline to keep running on its daily schedule.

  • The Cloud Scheduler job fires once daily. Each execution uses well under 1 minute of Cloud Run time, staying within the free tier (240,000 vCPU-seconds/month).
  • BigQuery storage is free for the first 10 GB. A single-tenant daily pull produces negligible data volume.
  • Cloud Scheduler allows 3 free jobs per month. Your single job counts as one.
  • The service account key remains on your machine. Keep it secure and do not commit it to version control.

Pause - I'll come back to this later

Stop automatic runs to free up your Cloud Scheduler free-tier slot. The Cloud Run job and BigQuery data remain but cost nothing when idle.

  • Delete the Cloud Scheduler job by running:
gcloud scheduler jobs delete workspace-usage-daily --location us-central1
  • Confirm with Y when prompted.
  • Your Cloud Run job, BigQuery dataset, and local files stay intact. When you are ready to resume, recreate the Cloud Scheduler job using the same command from Step 5.

Delete - I don't want to use this again

Remove all project resources and start fresh. Run each command below and confirm when prompted.

  • Delete the Cloud Scheduler job:
gcloud scheduler jobs delete workspace-usage-daily --location us-central1
  • Delete the Cloud Run job:
gcloud run jobs delete workspace-usage-pipeline --region us-central1
  • Delete the BigQuery dataset and all its tables:
bq rm -r workspace_usage

What does bq rm -r do?

The -r flag removes all tables inside the dataset before deleting the dataset itself. Without it, the command fails if any tables exist.

  • Delete the service account:
gcloud iam service-accounts delete workspace-usage-pipeline@[[GCP_PROJECT_ID="your-project-id"]].iam.gserviceaccount.com
  • Delete the local JSON key file:
rm workspace-usage-pipeline-key.json
  • Delete your local project directory (adjust the path if you saved it elsewhere):
deactivate
cd ~/Desktop
rm -rf workspace-usage-pipeline
  • Optionally, ask your Workspace admin to remove the domain-wide delegation entry for your service account's Client ID. In the Google Admin Console, they navigate to Security, then Access and data control, then API controls, then Manage Domain Wide Delegation, and delete the entry matching your Client ID.

Nice Work!

Nice Work!

You just built and deployed a production-ready data ingestion pipeline that pulls Google Workspace usage metrics from the Admin SDK Reports API and loads them into BigQuery on a daily schedule.

You've learned how to:

  • Build a production data pipeline from API to warehouse, using the architecture pattern (Cloud Scheduler > Cloud Run job > API > BigQuery) that senior data engineers use for recurring batch ingestion.
  • Design an idempotent, lag-aware pipeline using WRITE_TRUNCATE at the partition level and date windowing logic that accounts for Google's data availability delays.
  • Implement domain-wide delegation with a single read-only scope, following the principle of least privilege to securely access the Admin SDK Reports API.
  • Secret Mission: Made the pipeline config-driven so onboarding a new client is a config entry, not a code change.

Ready to quiz yourself?