Deploy a Workspace Metrics Pipeline

Build a config-driven pipeline that pulls Workspace metrics into BigQuery.

Introduction

30 Second Summary

Every team wants to know how their collaboration tools are actually being used. Are people adopting the apps the company pays for, or are licenses going to waste?

In this project, you will build a production-grade data pipeline that pulls Google Workspace usage metrics, aggregates them by department without storing any personal data, and loads them into BigQuery on an automated daily schedule.

What You'll Build

A config-driven Python pipeline deployed to Google Cloud that ingests Workspace adoption metrics from the Admin SDK Reports API, aggregates away PII in memory, and writes idempotently to a date-partitioned BigQuery table.

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

  • A config-driven pipeline that pulls per-department usage metrics from the Admin SDK Reports API without storing any user-level data.
  • A date-partitioned BigQuery table with idempotent writes so re-runs and backfills never double-count.
  • A Cloud Run job triggered by Cloud Scheduler on a daily cron, deployed with no credential files shipped in the image.
  • Secret Mission: Add a historical backfill mode with date-range CLI arguments and rate-limit-aware iteration over months of data.

Are there any prerequisites?

This project assumes you have a GCP project with billing enabled, the gcloud CLI installed, and Python 3.x on your machine. You also need access to a Workspace admin who can grant domain-wide delegation for the service account you create in Step 1.

Before We Start

Before We Start...

Set Up GCP Infrastructure and Service Account

Before you can pull Workspace metrics and load them into a data warehouse, the infrastructure needs to be in place. Your pipeline depends on five GCP APIs, a service account with domain-wide delegation, and a BigQuery table designed for idempotent writes.

In this step you will wire up the permissions layer and storage layer so that everything is ready when you write code in the next step.

In this step, get ready to:
  • Enable all required GCP APIs and create a least-privilege service account.
  • Configure domain-wide delegation so the service account can read Workspace usage data.
  • Create a date-partitioned BigQuery table with a long/narrow schema.
Enable APIs and create the service account

Your pipeline talks to several Google Cloud services. Each service has its own API that must be explicitly enabled before your project can call it.

  • Enable all five APIs at once by running this command in your terminal:
gcloud services enable \
  admin.googleapis.com \
  bigquery.googleapis.com \
  run.googleapis.com \
  cloudscheduler.googleapis.com \
  artifactregistry.googleapis.com

What does this command do?

Each service name (like admin.googleapis.com) is the API identifier for a GCP service.

  • admin.googleapis.com enables the Admin SDK Reports API so you can pull Workspace usage data.
  • bigquery.googleapis.com enables BigQuery for data warehousing.
  • run.googleapis.com enables Cloud Run for deploying the pipeline as a job.
  • cloudscheduler.googleapis.com enables Cloud Scheduler for triggering the job on a cron.
  • artifactregistry.googleapis.com enables Artifact Registry for storing the container image that Cloud Run builds from your source.
  • You should see each API being enabled in the output. This may take 30-60 seconds.

Seeing a permission denied error?

Your account needs the Service Usage Admin role (roles/serviceusage.serviceUsageAdmin) or Owner role to enable APIs. If you are not the project owner, ask your admin to grant this role.

Also confirm your gcloud CLI is pointed at the correct project by running gcloud config get-value project.

help me fix the permission denied error when enabling GCP APIs

Now create the service account that will run your pipeline. This account gets the minimum permissions it needs and nothing more.

  • Create the service account by running:
gcloud iam service-accounts create workspace-metrics-sa \
  --display-name="Workspace Metrics Pipeline"

What is a service account?

A service account is a special Google account that belongs to your application, not to a person. Your pipeline will authenticate as this service account when it calls the Admin SDK and BigQuery APIs.

The name workspace-metrics-sa generates an email like workspace-metrics-sa@your-project.iam.gserviceaccount.com. You will use this email when granting roles.

  • Confirm the service account was created by running:
gcloud iam service-accounts list
  • You should see Workspace Metrics Pipeline in the output list.

Next, you need to enable domain-wide delegation on this service account. This is done in the GCP Console, not the CLI.

  • Open the Google Cloud Console in your browser and navigate to IAM & Admin in the left sidebar.
  • Click Service Accounts.
  • Click on your workspace-metrics-sa service account.
  • Click the Advanced settings section to expand it.
  • Under Domain-wide delegation, check the box to Enable Google Workspace Domain-wide Delegation.
  • Click Save.

You will also need the service account's client ID for the next substep.

  • Retrieve the client ID by running:
gcloud iam service-accounts describe \
  workspace-metrics-sa@[[GCP_PROJECT="your-gcp-project-id"]].iam.gserviceaccount.com \
  --format="value(oauth2ClientId)"
  • Copy the numeric client ID from the output. You will paste it in the next substep.

Why domain-wide delegation?

The Admin SDK Reports API requires a Workspace admin identity to read usage data. Domain-wide delegation lets your service account impersonate a Workspace admin without storing any admin credentials.

You will restrict it to the single read-only scope https://www.googleapis.com/auth/admin.reports.usage.readonly so the service account can only read usage reports, not modify anything.

Grant domain-wide delegation in the Admin Console

Enabling delegation in GCP is only half the configuration. A Workspace admin must also authorize the service account's client ID in the Google Admin Console.

  • Provide your Workspace admin (or open it yourself if you are the admin) with the following instructions.
  • Open the Google Admin Console at admin.google.com.
  • Navigate to Security in the left menu.
  • Click Access and data control.
  • Click API controls.
  • Click Domain wide delegation.
  • Click Manage Domain Wide Delegation.
  • Click Add new.
  • In the Client ID field, paste the numeric client ID you copied earlier.
  • In the OAuth scopes field, enter exactly:
https://www.googleapis.com/auth/admin.reports.usage.readonly
  • Click Authorize.

Why only one scope?

This single scope is the minimum needed for the Admin SDK User Usage Reports endpoint. Adding more scopes would let the service account access data it does not need, violating the principle of least privilege.

Don't have access to the Admin Console?

Domain-wide delegation must be authorized by a Workspace super admin. If you are not the admin, send them the client ID and the scope string above. They only need to complete the steps in the Admin Console.

help me get domain-wide delegation authorized when I'm not the Workspace admin

Create the BigQuery dataset and date-partitioned table

The pipeline needs a destination to write aggregated metrics. You will create a dataset and a table with a long/narrow schema designed so new metrics never require schema changes.

  • Create the workspace_metrics dataset by running:
bq mk --dataset workspace_metrics

What is a BigQuery dataset?

A dataset is a top-level container in BigQuery that holds tables and views. It controls access permissions and data location.

  • Verify the dataset was created by running:
bq ls
  • You should see workspace_metrics listed in the output.

Now grant the service account permission to write data into this dataset.

  • Grant the BigQuery Data Editor role to your service account on the dataset by running:
bq add-iam-policy-binding \
  --member="serviceAccount:workspace-metrics-sa@[[GCP_PROJECT="your-gcp-project-id"]].iam.gserviceaccount.com" \
  --role="roles/bigquery.dataEditor" \
  [[GCP_PROJECT="your-gcp-project-id"]]:workspace_metrics

Why dataset-level instead of project-level?

Granting BigQuery Data Editor on the specific dataset instead of the whole project follows least-privilege. The service account can only read/write tables in workspace_metrics, not any other dataset in your project.

Now create the adoption_metrics table with daily partitioning on the report_date column.

  • Create the date-partitioned table by running:
bq mk \
  --table \
  --time_partitioning_field report_date \
  --time_partitioning_type DAY \
  workspace_metrics.adoption_metrics \
  report_date:DATE,client_id:STRING,segment:STRING,app_name:STRING,metric_name:STRING,metric_value:FLOAT64,loaded_at:TIMESTAMP

Why this schema shape?

This is called a long/narrow schema. Instead of one column per metric (which forces schema changes whenever Google adds a new metric), you store each metric as a row with metric_name and metric_value.

Queries become simple WHERE metric_name = 'gmail:num_emails_sent' filters. New metrics appear automatically without an ALTER TABLE.

Daily partitioning on report_date means the pipeline can use partition decorators and WRITE_TRUNCATE to overwrite a single day atomically. Re-runs never double-count.

  • Verify the table was created by running:
bq show workspace_metrics.adoption_metrics
  • You should see the table schema with all seven columns listed and the Time Partitioning field showing report_date with type DAY.

Seeing 'dataset not found' or 'already exists'?

If the dataset was not found, check that gcloud is configured with the correct project. Run gcloud config get-value project to confirm.

If the table already exists from a previous attempt, you can safely delete it with bq rm -f workspace_metrics.adoption_metrics and re-run the create command.

help me fix my BigQuery dataset or table creation error

Your infrastructure is ready. Next up, you will write a proof-of-concept script that authenticates as this service account and pulls raw usage data from the Admin SDK.

Prove Access Locally

Your GCP infrastructure is ready. You have a service account with domain-wide delegation, a BigQuery table waiting for data, and all the right APIs enabled.

Before building the full pipeline, you need to prove that the delegation chain actually works. In this step, you will write a small proof-of-concept script that authenticates as your service account, pulls one day of Workspace usage data, and prints the raw response so you can understand its nested structure.

In this step, get ready to:
  • Write a proof-of-concept script that authenticates via domain-wide delegation and calls the User Usage Report endpoint.
  • Handle pagination and inspect API warnings for data availability.
  • Identify PII fields in the response and plan the aggregation strategy.
Write the proof-of-concept script

You need a small script that exercises the full authentication chain: your local credentials impersonate the service account, which then impersonates the Workspace admin to call the Reports API. If any link in that chain is misconfigured, this script will surface the error immediately.

  • Install the Python packages you need for this script by running:
pip install google-api-python-client google-auth google-cloud-bigquery pyyaml

What do these packages do?

  • google-api-python-client provides the client for calling the Admin SDK Reports API.
  • google-auth handles authentication, including Application Default Credentials and domain-wide delegation.
  • google-cloud-bigquery is the BigQuery client you will use in later steps.
  • pyyaml parses YAML config files (needed in the next step).
  • Set up Application Default Credentials to impersonate your service account locally by running:
gcloud auth application-default login --impersonate-service-account=[[SA_EMAIL="workspace-metrics-sa@your-project.iam.gserviceaccount.com"]]

Why impersonate the service account?

Domain-wide delegation is granted to the service account, not to your personal Google account. By impersonating the service account locally, your Python script's default() call picks up the service account identity. The delegation chain then works end-to-end without downloading a key file.

  • A browser window will open asking you to authorize. Sign in with your Google account and allow the impersonation.
  • Create a new file called proof_of_concept.py in your project folder. Add the following imports, constants, and authentication function:
import json
import logging
from datetime import date, timedelta

from google.auth import default
from googleapiclient.discovery import build

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(message)s",
)
logger = logging.getLogger(__name__)

# The read-only scope for usage reports
SCOPES = ["https://www.googleapis.com/auth/admin.reports.usage.readonly"]

# Replace with your Workspace admin email
ADMIN_EMAIL = "[[ADMIN_EMAIL_VALUE=\"admin@yourcompany.com\"]]"

# Reports lag behind by approximately 5 days
DATA_LAG_DAYS = 5


def get_delegated_credentials(admin_email):
    """Authenticate with ADC and impersonate the admin via domain-wide delegation."""
    credentials, _ = default(scopes=SCOPES)
    delegated = credentials.with_subject(admin_email)
    return delegated

What does this code do?

  • default(scopes=SCOPES) loads the Application Default Credentials you just configured. Because you set up impersonation, these credentials represent your service account.
  • credentials.with_subject(admin_email) tells the service account to impersonate the Workspace admin. This is the domain-wide delegation step that lets the service account call the Reports API on behalf of the admin.
  • DATA_LAG_DAYS = 5 offsets the target date by 5 days. Google's usage reports take time to process, so requesting today's data returns empty results or warnings.
  • Below the get_delegated_credentials function, add the main() function that calls the User Usage Report endpoint and prints the raw JSON response:
def main():
    # Calculate the target date (5 days ago to avoid data lag)
    target_date = date.today() - timedelta(days=DATA_LAG_DAYS)
    report_date_str = target_date.isoformat()
    logger.info("Target date: %s", report_date_str)

    # Authenticate with domain-wide delegation
    credentials = get_delegated_credentials(ADMIN_EMAIL)

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

    # Call the User Usage Report endpoint for a single page
    response = (
        service.userUsageReport()
        .get(userKey="all", date=report_date_str)
        .execute()
    )

    # Print the raw JSON to inspect the nested structure
    reports = response.get("usageReports", [])
    logger.info("Fetched %d user records (first page)", len(reports))

    if reports:
        # Show the first user report to understand the structure
        print(json.dumps(reports[0], indent=2))


if __name__ == "__main__":
    main()

What does this code do?

  • build("admin", "reports_v1", credentials=credentials) creates a client for the Admin SDK Reports API v1.
  • userUsageReport().get(userKey="all", date=report_date_str) requests usage data for all users on the target date.
  • The response's usageReports array contains one entry per user. Each entry has an entity block (with userEmail and profileId) and a parameters array of metrics.
  • Save proof_of_concept.py.
  • Run the script to verify authentication works:
python proof_of_concept.py

You should see a log line showing the target date, a count of fetched records, and a JSON dump of the first user's report structure.

Seeing a permission or delegation error?

  • If you see a 403 error mentioning "Not Authorized to access this resource/api", double-check that domain-wide delegation in the Admin Console uses the service account's client ID (a numeric value), not the service account email.
  • If you see "invalid_grant", verify that ADMIN_EMAIL is a real Workspace admin address with access to the Reports API.
  • If you see "Data is not available", your target date is too recent. Increase DATA_LAG_DAYS to 7 and try again.

Still stuck? Help me debug my domain-wide delegation authentication error.

Handle pagination and inspect warnings

Large Workspace domains return hundreds of user records. The API paginates results and includes a nextPageToken field when more pages are available. Your script needs to loop until that token is absent.

  • In proof_of_concept.py, add a new function called fetch_user_usage above the main() function. This function handles pagination and warning inspection:
def fetch_user_usage(credentials, report_date_str):
    """Pull all user usage reports for a single date, handling pagination."""
    service = build("admin", "reports_v1", credentials=credentials)
    all_reports = []
    page_token = None
    page_count = 0

    while True:
        response = (
            service.userUsageReport()
            .get(userKey="all", date=report_date_str, pageToken=page_token)
            .execute()
        )

        # Check for data availability warnings
        warnings = response.get("warnings", [])
        for warning in warnings:
            msg = warning.get("message", "")
            logger.warning("API warning for %s: %s", report_date_str, msg)
            if "not available" in msg.lower():
                logger.error(
                    "Data not available for %s. Try an older date.", report_date_str
                )
                return None

        reports = response.get("usageReports", [])
        all_reports.extend(reports)
        page_count += 1
        logger.info(
            "Page %d: fetched %d records for %s",
            page_count, len(reports), report_date_str
        )

        # Follow nextPageToken until exhausted
        page_token = response.get("nextPageToken")
        if not page_token:
            break

    logger.info(
        "Total: %d user records across %d pages", len(all_reports), page_count
    )
    return all_reports

What does this code do?

  • The while True loop keeps fetching pages until nextPageToken is absent from the response, meaning all records have been returned.
  • The warnings check looks for messages containing "not available". The API returns these when the requested date is too recent and data processing is incomplete.
  • Returning None on a data-unavailable warning lets the caller distinguish between "no data exists" and "the API cannot provide it yet."
  • Now replace the existing main() function with this updated version that uses fetch_user_usage and inspects the response structure:
def main():
    # Calculate the target date (5 days ago to avoid data lag)
    target_date = date.today() - timedelta(days=DATA_LAG_DAYS)
    report_date_str = target_date.isoformat()
    logger.info("Target date: %s", report_date_str)

    # Authenticate with domain-wide delegation
    credentials = get_delegated_credentials(ADMIN_EMAIL)

    # Fetch all usage reports with pagination
    usage_reports = fetch_user_usage(credentials, report_date_str)

    if usage_reports is None:
        logger.error("No data returned. Exiting.")
        return

    if not usage_reports:
        logger.warning("Empty results. The date may be too recent.")
        return

    # Print the structure of the first report for inspection
    logger.info("--- Sample user report structure ---")
    print(json.dumps(usage_reports[0], indent=2))

    # Summarize entity and parameter fields
    sample = usage_reports[0]
    entity = sample.get("entity", {})
    logger.info("Entity fields: %s", list(entity.keys()))

    parameters = sample.get("parameters", [])
    logger.info("Number of parameters: %d", len(parameters))
    if parameters:
        logger.info("First parameter: %s", json.dumps(parameters[0], indent=2))


if __name__ == "__main__":
    main()

What does the updated main do?

  • It calls fetch_user_usage which handles all pagination automatically.
  • It handles the None return (data unavailable) and empty results separately so you get clear log messages either way.
  • It prints the entity fields and first parameter to help you understand the nested structure before building the real pipeline.
  • Save proof_of_concept.py and run it again:
python proof_of_concept.py

You should see log lines showing how many pages were fetched, the total user record count, and a JSON dump of a sample user report.

Seeing zero records or empty results?

  • If you see "Data not available", increase DATA_LAG_DAYS to 7 or 10. Some Workspace domains have longer processing delays.
  • If you see 0 user records with no warnings, confirm your admin email has Reports API access. Not all admin roles include Reports privileges.

My script runs but returns zero records. Help me figure out why.

✔️ Awesome, I've got everything!

Great. Double check you have saved proof_of_concept.py with the pagination and warning handling in place.

ⓧ I'd like to double check the full code

import json
import logging
from datetime import date, timedelta

from google.auth import default
from googleapiclient.discovery import build

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(message)s",
)
logger = logging.getLogger(__name__)

# The read-only scope for usage reports
SCOPES = ["https://www.googleapis.com/auth/admin.reports.usage.readonly"]

# Replace with your Workspace admin email
ADMIN_EMAIL = "admin@yourcompany.com"

# Reports lag behind by approximately 5 days
DATA_LAG_DAYS = 5


def get_delegated_credentials(admin_email):
    """Authenticate with ADC and impersonate the admin via domain-wide delegation."""
    credentials, _ = default(scopes=SCOPES)
    delegated = credentials.with_subject(admin_email)
    return delegated


def fetch_user_usage(credentials, report_date_str):
    """Pull all user usage reports for a single date, handling pagination."""
    service = build("admin", "reports_v1", credentials=credentials)
    all_reports = []
    page_token = None
    page_count = 0

    while True:
        response = (
            service.userUsageReport()
            .get(userKey="all", date=report_date_str, pageToken=page_token)
            .execute()
        )

        # Check for data availability warnings
        warnings = response.get("warnings", [])
        for warning in warnings:
            msg = warning.get("message", "")
            logger.warning("API warning for %s: %s", report_date_str, msg)
            if "not available" in msg.lower():
                logger.error(
                    "Data not available for %s. Try an older date.", report_date_str
                )
                return None

        reports = response.get("usageReports", [])
        all_reports.extend(reports)
        page_count += 1
        logger.info(
            "Page %d: fetched %d records for %s",
            page_count, len(reports), report_date_str
        )

        # Follow nextPageToken until exhausted
        page_token = response.get("nextPageToken")
        if not page_token:
            break

    logger.info(
        "Total: %d user records across %d pages", len(all_reports), page_count
    )
    return all_reports


def main():
    # Calculate the target date (5 days ago to avoid data lag)
    target_date = date.today() - timedelta(days=DATA_LAG_DAYS)
    report_date_str = target_date.isoformat()
    logger.info("Target date: %s", report_date_str)

    # Authenticate with domain-wide delegation
    credentials = get_delegated_credentials(ADMIN_EMAIL)

    # Fetch all usage reports with pagination
    usage_reports = fetch_user_usage(credentials, report_date_str)

    if usage_reports is None:
        logger.error("No data returned. Exiting.")
        return

    if not usage_reports:
        logger.warning("Empty results. The date may be too recent.")
        return

    # Print the structure of the first report for inspection
    logger.info("--- Sample user report structure ---")
    print(json.dumps(usage_reports[0], indent=2))

    # Summarize entity and parameter fields
    sample = usage_reports[0]
    entity = sample.get("entity", {})
    logger.info("Entity fields: %s", list(entity.keys()))

    parameters = sample.get("parameters", [])
    logger.info("Number of parameters: %d", len(parameters))
    if parameters:
        logger.info("First parameter: %s", json.dumps(parameters[0], indent=2))


if __name__ == "__main__":
    main()
Identify PII fields and plan aggregation

Look at the JSON output your script printed. Each user report has a predictable structure that you will exploit in the real pipeline.

What is the response structure?

Each user report contains two main sections:

  • entity block: contains userEmail, profileId, and customerId. This is the user's identity.
  • parameters array: each entry has a name (like gmail:num_emails_sent) and a typed value field (intValue, stringValue, boolValue, datetimeValue, or msgValue). These are the actual usage metrics.

The parameter name field uses a app_name:metric_name pattern. This colon-separated format is how you will split metrics into app and metric dimensions in the pipeline.

Which fields contain PII?

Three fields in the response identify individual users:

  • entity.userEmail is the user's full email address.
  • accounts:first_name and accounts:last_name appear as parameters in the response.

Your pipeline will never write these fields to BigQuery. Instead, it will use the email address only to match users to departments (via config), then aggregate the numeric metrics at the department level. The user identity exists only in memory during processing and is discarded before any data reaches the warehouse.

What is the aggregation strategy?

For each user, the pipeline will:

  • Look up the user's email in a config-driven department mapping to determine their segment (Engineering, Sales, Marketing, etc.).
  • Sum all numeric metric values at the segment level. One row per (date, segment, app, metric) goes to BigQuery.
  • Discard the user identity entirely. The final table contains zero user-level data.

This is the PII-in-memory-only pattern. It means even if your BigQuery table is compromised, no individual user data is exposed.

You have proven that the delegation chain works end-to-end and you understand the raw API response. Next up, you will build the real pipeline that reads from config, aggregates metrics by department, and loads them into BigQuery.

Build the Config-Driven Pipeline Core

Your proof-of-concept script confirmed that domain-wide delegation works and you can pull raw usage data from the Admin SDK. Now it is time to turn those learnings into a production-grade pipeline.

In this step you will build a config-driven, multi-tenant pipeline that extracts per-user Workspace metrics, aggregates them to the department level (so no user-level PII ever reaches BigQuery), and writes the results idempotently using partition decorators.

In this step, get ready to:
  • Create a tenant config file and pin your Python dependencies.
  • Build extraction, aggregation, and row-flattening functions in the pipeline script.
  • Add idempotent BigQuery writes and wire the main orchestrator function.
Create the config file and pin dependencies

A config-driven pipeline separates what you process from how you process it. Adding a new tenant means dropping a YAML block into the config file, never editing pipeline code.

  • Create a file called config.yaml in your project folder. Copy this tenant definition into it:
tenants:
  - tenant_id: "internal"
    admin_email: "admin@yourcompany.com"
    scopes:
      - "https://www.googleapis.com/auth/admin.reports.usage.readonly"
    department_mappings:
      "engineering@yourcompany.com": "Engineering"
      "sales@yourcompany.com": "Sales"
      "marketing@yourcompany.com": "Marketing"
      "*": "Other"
    bigquery:
      project: "your-gcp-project"
      dataset: "workspace_metrics"
      table: "adoption_metrics"

What does this config do?

Each entry under tenants defines one client or business unit the pipeline processes.

  • admin_email is the Workspace admin the service account impersonates via domain-wide delegation.
  • department_mappings maps email patterns to human-readable segment names. The wildcard "*" catches users that do not match any explicit pattern.
  • The bigquery block tells the pipeline exactly where to write results for this tenant.
  • Replace admin@yourcompany.com with the email address of your Workspace admin.
  • Replace your-gcp-project with your actual GCP project ID.
  • Update the department_mappings keys to match real email addresses or patterns in your Workspace domain.
  • Save config.yaml.

Why config-driven from day one?

Onboarding a second client means adding another block under tenants. The pipeline loops over every tenant automatically. You never touch the Python code to expand coverage.

  • Create a file called requirements.txt in the same folder and paste these pinned dependencies:
google-api-python-client==2.197.0
google-auth==2.53.0
google-cloud-bigquery==3.41.0
pyyaml==6.0.2

What are these packages?

  • google-api-python-client provides the Admin SDK client for calling the Reports API.
  • google-auth handles OAuth2 service account credentials and domain-wide delegation.
  • google-cloud-bigquery is the BigQuery Python client for loading data.
  • pyyaml parses your config.yaml file.
  • Save requirements.txt.
  • Install the dependencies by running:
pip install -r requirements.txt
  • Confirm both files exist by running:
ls

You should see config.yaml and requirements.txt listed in the output alongside your earlier proof_of_concept.py.

Pip install failing?

  • Make sure you are running the command from the same folder where requirements.txt lives.
  • If you see a permissions error, try pip install --user -r requirements.txt or use a virtual environment.

Help me debug a pip install failure for my requirements.txt

Build the extraction and transformation functions

The pipeline script combines everything you learned from the proof-of-concept into reusable functions. Each function handles one responsibility: authentication, API calls, segment mapping, aggregation, or row flattening.

  • Create a new file called pipeline.py and paste the following imports, logging setup, and helper functions:
import json
import logging
import os
import sys
from datetime import date, datetime, timedelta, timezone

import yaml
from google.auth import default
from google.oauth2 import service_account
from google.cloud import bigquery
from googleapiclient.discovery import build

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(message)s",
)
logger = logging.getLogger(__name__)

SCOPES = ["https://www.googleapis.com/auth/admin.reports.usage.readonly"]
DATA_LAG_DAYS = 5


def load_config(path="config.yaml"):
    with open(path, "r") as f:
        return yaml.safe_load(f)


def get_delegated_credentials(admin_email):
    credentials, _ = default(scopes=SCOPES)
    delegated = credentials.with_subject(admin_email)
    return delegated

What does this code do?

  • The logging setup gives you timestamped, structured logs that Cloud Run surfaces automatically.
  • SCOPES defines the single read-only scope for the Reports API.
  • DATA_LAG_DAYS = 5 accounts for Google's processing delay on usage data.
  • load_config() reads your config.yaml and returns it as a Python dict.
  • get_delegated_credentials() uses Application Default Credentials and impersonates the admin email via domain-wide delegation.

You will not see a change yet. The next function uses these credentials to pull data.

  • Below get_delegated_credentials, add the data extraction function that handles pagination and API warnings:
def fetch_user_usage(credentials, report_date_str):
    service = build("admin", "reports_v1", credentials=credentials)
    all_reports = []
    page_token = None

    while True:
        response = (
            service.userUsageReport()
            .get(userKey="all", date=report_date_str, pageToken=page_token)
            .execute()
        )

        warnings = response.get("warnings", [])
        for warning in warnings:
            msg = warning.get("message", "")
            if "not available" in msg.lower():
                logger.warning(
                    "Data not available for %s: %s", report_date_str, msg
                )
                return None

        reports = response.get("usageReports", [])
        all_reports.extend(reports)
        logger.info(
            "Fetched page with %d records for %s", len(reports), report_date_str
        )

        page_token = response.get("nextPageToken")
        if not page_token:
            break

    return all_reports

What does this function do?

This is the same pagination and warning logic you tested in your proof-of-concept, wrapped in a reusable function.

  • It follows nextPageToken until all user records are collected.
  • If the API returns a warning containing "not available", the function returns None so the caller can skip that date gracefully.
  • Save pipeline.py and confirm there are no syntax errors by running:
python -c "import pipeline; print('OK')"

You should see OK printed with no errors.

Seeing an ImportError?

  • Check that you ran pip install -r requirements.txt successfully in the previous substep.
  • Make sure you are running the command from the same directory where pipeline.py lives.

Help me fix an ImportError in my pipeline.py file

  • Below fetch_user_usage, add the segment mapping and aggregation functions. These are the core of the PII-in-memory-only pattern:
def map_user_to_segment(user_email, department_mappings):
    for pattern, segment in department_mappings.items():
        if pattern == "*":
            continue
        if user_email.startswith(pattern.split("@")[0] + "+") or user_email == pattern:
            return segment
        if user_email.endswith("@" + pattern.split("@")[1] if "@" in pattern else ""):
            return segment
    return department_mappings.get("*", "Unknown")


def aggregate_to_segments(usage_reports, department_mappings):
    segment_metrics = {}

    for report in usage_reports:
        entity = report.get("entity", {})
        user_email = entity.get("userEmail", "")

        segment = map_user_to_segment(user_email, department_mappings)

        parameters = report.get("parameters", [])
        for param in parameters:
            name = param.get("name", "")
            if ":" not in name:
                continue

            app_name, metric_name = name.split(":", 1)

            value = None
            if "intValue" in param:
                value = float(param["intValue"])
            elif "stringValue" in param and param["stringValue"].isdigit():
                value = float(param["stringValue"])
            else:
                continue

            key = (segment, app_name, metric_name)
            if key not in segment_metrics:
                segment_metrics[key] = 0.0
            segment_metrics[key] += value

    return segment_metrics

How does the PII-in-memory-only pattern work?

The user_email variable exists only inside the loop iteration. It is used to look up a segment name, then discarded. The aggregated output contains only segment-level totals. No user identifier is ever written to BigQuery.

The map_user_to_segment function matches each user's email against the patterns in your config.yaml. If no pattern matches, the wildcard "*" catches them as "Other".

You will not see a change yet. The next function transforms the aggregated dict into rows for BigQuery.

  • Below aggregate_to_segments, add the row builder that produces the long/narrow shape:
def build_rows(segment_metrics, report_date, tenant_id):
    loaded_at = datetime.now(timezone.utc).isoformat()
    rows = []

    for (segment, app_name, metric_name), metric_value in segment_metrics.items():
        rows.append(
            {
                "report_date": report_date.isoformat(),
                "client_id": tenant_id,
                "segment": segment,
                "app_name": app_name,
                "metric_name": metric_name,
                "metric_value": metric_value,
                "loaded_at": loaded_at,
            }
        )

    return rows

Why a long/narrow table shape?

Each row holds one metric for one segment on one date. When Google adds new metrics (they frequently do), the table absorbs them without any schema changes.

A wide table with one column per metric would require ALTER TABLE every time a new metric appears. The long/narrow shape trades slightly more verbose queries for zero operational burden.

  • Save pipeline.py and confirm the syntax is still clean:
python -c "import pipeline; print('OK')"

You should see OK printed again with no tracebacks.

Add idempotent writes and wire the main orchestrator

The final piece is the BigQuery write function and the main loop that ties everything together. The write uses a partition decorator with WRITE_TRUNCATE so re-running for the same date atomically replaces that date's partition without touching other data.

  • Below build_rows, add the BigQuery load function:
def load_to_bigquery(rows, project, dataset, table, report_date):
    client = bigquery.Client(project=project)

    partition_date = report_date.strftime("%Y%m%d")
    table_id = f"{project}.{dataset}.{table}${partition_date}"

    job_config = bigquery.LoadJobConfig(
        write_disposition=bigquery.WriteDisposition.WRITE_TRUNCATE,
        source_format=bigquery.SourceFormat.NEWLINE_DELIMITED_JSON,
        schema=[
            bigquery.SchemaField("report_date", "DATE"),
            bigquery.SchemaField("client_id", "STRING"),
            bigquery.SchemaField("segment", "STRING"),
            bigquery.SchemaField("app_name", "STRING"),
            bigquery.SchemaField("metric_name", "STRING"),
            bigquery.SchemaField("metric_value", "FLOAT64"),
            bigquery.SchemaField("loaded_at", "TIMESTAMP"),
        ],
    )

    load_job = client.load_table_from_json(rows, table_id, job_config=job_config)
    load_job.result()

    logger.info(
        "Loaded %d rows into %s (partition %s)", len(rows), table_id, partition_date
    )

What makes this idempotent?

The partition decorator table$YYYYMMDD targets a single date's partition. WRITE_TRUNCATE replaces all rows in that partition atomically.

If the job fails mid-way and retries, or if you backfill the same date twice, the result is always correct. Other date partitions are never touched. This is simpler than a MERGE statement and avoids full-table scans.

You will not see a change yet. The orchestrator function below calls load_to_bigquery after aggregation.

  • Below load_to_bigquery, add the date computation, main orchestrator, and entrypoint:
def get_target_dates():
    start_date_str = os.environ.get("BACKFILL_START_DATE")
    end_date_str = os.environ.get("BACKFILL_END_DATE")

    if start_date_str and end_date_str:
        start = date.fromisoformat(start_date_str)
        end = date.fromisoformat(end_date_str)
        dates = []
        current = start
        while current <= end:
            dates.append(current)
            current += timedelta(days=1)
        logger.info("Backfill mode: %d dates from %s to %s", len(dates), start, end)
        return dates

    target = date.today() - timedelta(days=DATA_LAG_DAYS)
    return [target]


def run_pipeline():
    config = load_config()

    target_dates = get_target_dates()

    for tenant in config["tenants"]:
        tenant_id = tenant["tenant_id"]
        admin_email = tenant["admin_email"]
        department_mappings = tenant["department_mappings"]
        bq_config = tenant["bigquery"]

        logger.info("Processing tenant: %s", tenant_id)

        credentials = get_delegated_credentials(admin_email)

        for report_date in target_dates:
            report_date_str = report_date.isoformat()
            logger.info("Fetching data for %s / %s", tenant_id, report_date_str)

            usage_reports = fetch_user_usage(credentials, report_date_str)

            if usage_reports is None:
                logger.warning(
                    "Skipping %s for %s - data not available", report_date_str, tenant_id
                )
                continue

            if not usage_reports:
                logger.warning(
                    "No usage reports returned for %s / %s", tenant_id, report_date_str
                )
                continue

            segment_metrics = aggregate_to_segments(usage_reports, department_mappings)
            rows = build_rows(segment_metrics, report_date, tenant_id)

            if rows:
                load_to_bigquery(
                    rows,
                    bq_config["project"],
                    bq_config["dataset"],
                    bq_config["table"],
                    report_date,
                )
            else:
                logger.info("No numeric metrics to load for %s / %s", tenant_id, report_date_str)

    logger.info("Pipeline complete.")


if __name__ == "__main__":
    run_pipeline()

What does the orchestrator do?

  • get_target_dates() defaults to 5 days ago (respecting data lag). If BACKFILL_START_DATE and BACKFILL_END_DATE environment variables are set, it generates every date in that range instead.
  • run_pipeline() reads the config, loops over each tenant, pulls data for each target date, aggregates it, and loads the results.
  • If a date's data is not yet available, the pipeline logs a warning and moves on. No empty data is written.
  • Save pipeline.py.
  • Run the full pipeline locally to verify end-to-end:
python pipeline.py

You should see structured log output showing the tenant being processed, the date being fetched, and either rows loaded into BigQuery or a warning that data is not yet available.

Seeing a permissions or authentication error?

  • Make sure your gcloud CLI is authenticated with Application Default Credentials. Run gcloud auth application-default login if needed.
  • Confirm that the admin_email in your config.yaml matches the admin account granted in the domain-wide delegation setup.
  • Double-check that your service account's client ID is registered in the Google Admin Console under Manage Domain Wide Delegation with the correct scope.

Help me fix an authentication or permissions error in my Workspace metrics pipeline

✔️ Awesome, I've got everything!

Great. Double check you have saved all three files: config.yaml, requirements.txt, and pipeline.py.

ⓧ I'd like to double check the full code

Here is the complete pipeline.py for comparison:

import json
import logging
import os
import sys
from datetime import date, datetime, timedelta, timezone

import yaml
from google.auth import default
from google.oauth2 import service_account
from google.cloud import bigquery
from googleapiclient.discovery import build

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(message)s",
)
logger = logging.getLogger(__name__)

SCOPES = ["https://www.googleapis.com/auth/admin.reports.usage.readonly"]
DATA_LAG_DAYS = 5


def load_config(path="config.yaml"):
    with open(path, "r") as f:
        return yaml.safe_load(f)


def get_delegated_credentials(admin_email):
    credentials, _ = default(scopes=SCOPES)
    delegated = credentials.with_subject(admin_email)
    return delegated


def fetch_user_usage(credentials, report_date_str):
    service = build("admin", "reports_v1", credentials=credentials)
    all_reports = []
    page_token = None

    while True:
        response = (
            service.userUsageReport()
            .get(userKey="all", date=report_date_str, pageToken=page_token)
            .execute()
        )

        warnings = response.get("warnings", [])
        for warning in warnings:
            msg = warning.get("message", "")
            if "not available" in msg.lower():
                logger.warning(
                    "Data not available for %s: %s", report_date_str, msg
                )
                return None

        reports = response.get("usageReports", [])
        all_reports.extend(reports)
        logger.info(
            "Fetched page with %d records for %s", len(reports), report_date_str
        )

        page_token = response.get("nextPageToken")
        if not page_token:
            break

    return all_reports


def map_user_to_segment(user_email, department_mappings):
    for pattern, segment in department_mappings.items():
        if pattern == "*":
            continue
        if user_email.startswith(pattern.split("@")[0] + "+") or user_email == pattern:
            return segment
        if user_email.endswith("@" + pattern.split("@")[1] if "@" in pattern else ""):
            return segment
    return department_mappings.get("*", "Unknown")


def aggregate_to_segments(usage_reports, department_mappings):
    segment_metrics = {}

    for report in usage_reports:
        entity = report.get("entity", {})
        user_email = entity.get("userEmail", "")

        segment = map_user_to_segment(user_email, department_mappings)

        parameters = report.get("parameters", [])
        for param in parameters:
            name = param.get("name", "")
            if ":" not in name:
                continue

            app_name, metric_name = name.split(":", 1)

            value = None
            if "intValue" in param:
                value = float(param["intValue"])
            elif "stringValue" in param and param["stringValue"].isdigit():
                value = float(param["stringValue"])
            else:
                continue

            key = (segment, app_name, metric_name)
            if key not in segment_metrics:
                segment_metrics[key] = 0.0
            segment_metrics[key] += value

    return segment_metrics


def build_rows(segment_metrics, report_date, tenant_id):
    loaded_at = datetime.now(timezone.utc).isoformat()
    rows = []

    for (segment, app_name, metric_name), metric_value in segment_metrics.items():
        rows.append(
            {
                "report_date": report_date.isoformat(),
                "client_id": tenant_id,
                "segment": segment,
                "app_name": app_name,
                "metric_name": metric_name,
                "metric_value": metric_value,
                "loaded_at": loaded_at,
            }
        )

    return rows


def load_to_bigquery(rows, project, dataset, table, report_date):
    client = bigquery.Client(project=project)

    partition_date = report_date.strftime("%Y%m%d")
    table_id = f"{project}.{dataset}.{table}${partition_date}"

    job_config = bigquery.LoadJobConfig(
        write_disposition=bigquery.WriteDisposition.WRITE_TRUNCATE,
        source_format=bigquery.SourceFormat.NEWLINE_DELIMITED_JSON,
        schema=[
            bigquery.SchemaField("report_date", "DATE"),
            bigquery.SchemaField("client_id", "STRING"),
            bigquery.SchemaField("segment", "STRING"),
            bigquery.SchemaField("app_name", "STRING"),
            bigquery.SchemaField("metric_name", "STRING"),
            bigquery.SchemaField("metric_value", "FLOAT64"),
            bigquery.SchemaField("loaded_at", "TIMESTAMP"),
        ],
    )

    load_job = client.load_table_from_json(rows, table_id, job_config=job_config)
    load_job.result()

    logger.info(
        "Loaded %d rows into %s (partition %s)", len(rows), table_id, partition_date
    )


def get_target_dates():
    start_date_str = os.environ.get("BACKFILL_START_DATE")
    end_date_str = os.environ.get("BACKFILL_END_DATE")

    if start_date_str and end_date_str:
        start = date.fromisoformat(start_date_str)
        end = date.fromisoformat(end_date_str)
        dates = []
        current = start
        while current <= end:
            dates.append(current)
            current += timedelta(days=1)
        logger.info("Backfill mode: %d dates from %s to %s", len(dates), start, end)
        return dates

    target = date.today() - timedelta(days=DATA_LAG_DAYS)
    return [target]


def run_pipeline():
    config = load_config()

    target_dates = get_target_dates()

    for tenant in config["tenants"]:
        tenant_id = tenant["tenant_id"]
        admin_email = tenant["admin_email"]
        department_mappings = tenant["department_mappings"]
        bq_config = tenant["bigquery"]

        logger.info("Processing tenant: %s", tenant_id)

        credentials = get_delegated_credentials(admin_email)

        for report_date in target_dates:
            report_date_str = report_date.isoformat()
            logger.info("Fetching data for %s / %s", tenant_id, report_date_str)

            usage_reports = fetch_user_usage(credentials, report_date_str)

            if usage_reports is None:
                logger.warning(
                    "Skipping %s for %s - data not available", report_date_str, tenant_id
                )
                continue

            if not usage_reports:
                logger.warning(
                    "No usage reports returned for %s / %s", tenant_id, report_date_str
                )
                continue

            segment_metrics = aggregate_to_segments(usage_reports, department_mappings)
            rows = build_rows(segment_metrics, report_date, tenant_id)

            if rows:
                load_to_bigquery(
                    rows,
                    bq_config["project"],
                    bq_config["dataset"],
                    bq_config["table"],
                    report_date,
                )
            else:
                logger.info("No numeric metrics to load for %s / %s", tenant_id, report_date_str)

    logger.info("Pipeline complete.")


if __name__ == "__main__":
    run_pipeline()

Your pipeline is working end-to-end locally. Next up, you will deploy it to Cloud Run and wire a Cloud Scheduler trigger so it runs automatically every day.

Deploy to Cloud Run and Schedule

Your pipeline pulls Workspace metrics, aggregates them to department level, and writes idempotent rows into BigQuery. All of that works locally.

Now it is time to move this pipeline into the cloud so it runs automatically every day. You will deploy it as a Cloud Run job triggered by Cloud Scheduler with no Docker knowledge and no credential files shipped in the image.

In this step, get ready to:
  • Create the Procfile so Cloud Run knows how to start your pipeline.
  • Deploy the Cloud Run job from source and verify it end-to-end.
  • Create a Cloud Scheduler trigger for automated daily execution.
Create the Procfile for deployment

Cloud Run needs two things to deploy your Python app from source without a Dockerfile. First, a requirements.txt so Cloud Buildpacks can detect it is a Python app. You already have that file. Second, a Procfile that tells the buildpack which command to run when the container starts.

  • Create a file called Procfile (no file extension) in the same directory as pipeline.py by running:
echo 'web: python pipeline.py' > Procfile
  • Confirm the file was created by running:
cat Procfile
  • You should see web: python pipeline.py printed in the terminal.

Why no Dockerfile?

When you run gcloud run jobs deploy --source ., GCP uses Cloud Buildpacks to detect your Python app via requirements.txt and build a container image automatically. The Procfile tells that container what command to execute. No Dockerfile, no Docker knowledge required.

Why no service account key file?

The --service-account flag on the deploy command attaches the service account identity directly to the Cloud Run job at the infrastructure level. Your Python code uses Application Default Credentials, which automatically pick up the attached identity at runtime.

This means no JSON key file is baked into the container image. No credential file means zero risk of credential leaks.

Deploy the Cloud Run job and verify end-to-end

With all deployment files in place, you can deploy directly from source. The gcloud CLI uploads your source code, builds the container in the cloud, and registers it as a Cloud Run job.

  • Deploy the Cloud Run job from source by running:
gcloud run jobs deploy workspace-metrics-pipeline \
  --source . \
  --region [[GCP_REGION="your GCP region e.g. us-central1"]] \
  --service-account [[SA_EMAIL="your-service-account@your-project.iam.gserviceaccount.com"]] \
  --project [[PROJECT_ID="your-gcp-project-id"]]

What does this command do?

  • --source . tells Cloud Run to upload the current directory, detect Python via the buildpack, and build a container.
  • --service-account attaches your service account (with domain-wide delegation and BigQuery Data Editor) as the job's runtime identity.
  • --project ensures the job deploys into the correct GCP project.
  • If prompted to enable the Artifact Registry API or create a repository, enter y to accept.
  • Wait for the deployment to finish. You should see a success message confirming the job was deployed.

Deployment failed?

  • Check that requirements.txt and Procfile are both in the same directory you are running the command from.
  • Verify the Artifact Registry API is enabled. The deploy command should prompt you, but you can enable it manually with gcloud services enable artifactregistry.googleapis.com.
  • Confirm your service account email is correct. You can find it in the GCP Console under IAM & Admin > Service Accounts.

Help me troubleshoot my Cloud Run deployment failure.

  • Execute the job manually to verify it works end-to-end by running:
gcloud run jobs execute workspace-metrics-pipeline \
  --region [[GCP_REGION="your GCP region e.g. us-central1"]]
  • Wait for the execution to complete. You should see a message indicating the execution finished successfully.
  • Verify data landed in BigQuery by running:
bq query --use_legacy_sql=false \
  'SELECT report_date, segment, app_name, COUNT(*) as row_count
   FROM `[[PROJECT_ID="your-gcp-project-id"]].workspace_metrics.adoption_metrics`
   GROUP BY report_date, segment, app_name
   LIMIT 10'
  • You should see rows with your report date, segment names from your config, and metric counts. This confirms the full pipeline ran successfully in the cloud.

Execution succeeded but no rows in BigQuery?

  • Check the Cloud Run job logs in the GCP Console under Cloud Run > Jobs > workspace-metrics-pipeline > Logs. Look for "Data not available" warnings, which mean the report date is too recent.
  • Verify the domain-wide delegation scope and admin email in config.yaml match what was configured in the Admin Console.
  • Confirm the service account has BigQuery Data Editor on the workspace_metrics dataset.

Help me debug why my pipeline execution shows no data in BigQuery.

Create the Cloud Scheduler trigger

The pipeline works on demand. Now you will set it up to run automatically every day. Cloud Scheduler sends an authenticated HTTP POST to the Cloud Run API endpoint that triggers your job.

Because the target URL is on *.googleapis.com, Cloud Scheduler uses an OAuth token (not an OIDC token) for authentication. The service account generating that token needs the Cloud Run Invoker role on your job.

  • First, grant the default Compute Engine service account the Cloud Run Invoker role on your job by running:
gcloud run jobs add-iam-policy-binding workspace-metrics-pipeline \
  --region [[GCP_REGION="your GCP region e.g. us-central1"]] \
  --member=serviceAccount:[[PROJECT_NUMBER="your-project-number"]]-compute@developer.gserviceaccount.com \
  --role=roles/run.invoker

Where do I find my project number?

In the GCP Console, go to the project Dashboard page. Your project number is displayed on the Project info card. It is a numeric ID (different from your project ID string).

  • Create the Cloud Scheduler job to trigger the pipeline daily at 6:00 AM by running:
gcloud scheduler jobs create http workspace-metrics-daily \
  --location [[GCP_REGION="your GCP region e.g. us-central1"]] \
  --schedule="0 6 * * *" \
  --uri="https://run.googleapis.com/v2/projects/[[PROJECT_ID="your-gcp-project-id"]]/locations/[[GCP_REGION="your GCP region e.g. us-central1"]]/jobs/workspace-metrics-pipeline:run" \
  --http-method POST \
  --oauth-service-account-email [[PROJECT_NUMBER="your-project-number"]]-compute@developer.gserviceaccount.com

What does this command do?

  • --schedule="0 6 * * *" is a cron expression meaning "every day at 6:00 AM".
  • --uri points to the Cloud Run Admin API v2 endpoint for your job. The :run suffix tells the API to execute the job.
  • --http-method POST sends a POST request, which is what the API requires to trigger an execution.
  • --oauth-service-account-email specifies which service account generates the OAuth token for authenticating the request to the Cloud Run API.
  • Verify the scheduler job was created by running:
gcloud scheduler jobs list --location [[GCP_REGION="your GCP region e.g. us-central1"]]
  • You should see workspace-metrics-daily in the output with a schedule of 0 6 * * * and a state of ENABLED.

Scheduler job creation failed?

  • Make sure the Cloud Scheduler API is enabled. Run gcloud services enable cloudscheduler.googleapis.com if needed.
  • Verify your project number is correct in the --oauth-service-account-email flag. It should be the numeric project number, not the project ID string.
  • Confirm the --uri contains your actual project ID and the correct region where the Cloud Run job is deployed.

Help me troubleshoot my Cloud Scheduler job creation.

✔️ Awesome, I've got everything!

Great. Your pipeline now runs automatically every day at 6:00 AM, pulling the latest Workspace metrics and loading them into BigQuery.

ⓧ I'd like to double check the full Procfile

Here is the complete Procfile:

web: python pipeline.py

Your pipeline is now fully deployed and scheduled. Every morning, Cloud Scheduler triggers the Cloud Run job, which pulls the latest Workspace usage data, aggregates it by department, and loads it idempotently into BigQuery. Next up, you will tackle the Secret Mission and add historical backfill capabilities.

Secret mission

Add Historical Backfill Mode

Your pipeline runs daily, but what about all the historical data you missed? Add CLI argument parsing so you can backfill months of Workspace metrics in a single command, with rate-limit-aware delays between API calls to stay under quota.

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 runs within GCP free tier limits, so there are no ongoing costs for the resources as configured.

Resources you used:

  • GCP service account with domain-wide delegation.
  • BigQuery dataset workspace_metrics with table adoption_metrics.
  • Cloud Run job (deployed from source).
  • Cloud Scheduler job (daily cron trigger).
  • Domain-wide delegation entry in Google Admin Console.
  • Local project files (pipeline.py, config.yaml, requirements.txt, Procfile).

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.

  • Cloud Scheduler triggers daily at no cost within free tier.
  • BigQuery storage is negligible for aggregated metrics data.
  • The Cloud Run job only incurs cost when executing, and runs well within the free tier allowance.

Pause - I'll come back to this later

Stop the daily scheduled runs to free up your Cloud Scheduler slot, but keep the job definition and data intact for later.

  • Delete the Cloud Scheduler job to stop automated runs by running:
gcloud scheduler jobs delete [[SCHEDULER_JOB_NAME="your-scheduler-job-name"]]
  • The Cloud Run job definition remains but costs nothing while idle.
  • Your BigQuery data stays in place for future queries.

When you are ready to resume, recreate the Cloud Scheduler job with the same cron schedule and URI.

Delete - I don't want to use this again

Remove all project resources and start fresh. Run each of the following commands in your terminal.

  • Delete the Cloud Scheduler job:
gcloud scheduler jobs delete [[SCHEDULER_JOB_NAME="your-scheduler-job-name"]]
  • Delete the Cloud Run job:
gcloud run jobs delete [[CLOUD_RUN_JOB_NAME="your-cloud-run-job-name"]]
  • Delete the BigQuery dataset and all its tables:
bq rm -r -f workspace_metrics
  • Delete the service account:
gcloud iam service-accounts delete [[SA_EMAIL="your-sa-name@your-project.iam.gserviceaccount.com"]]
  • Remove the domain-wide delegation entry. Ask your Workspace admin to navigate to Security > Access and data control > API controls > Domain wide delegation > Manage Domain Wide Delegation in the Google Admin Console and remove the entry for your service account's client ID.
  • Delete your local project files:
rm pipeline.py config.yaml requirements.txt Procfile

That's a Wrap!

That's a Wrap!

Nice work! You just built and deployed a production-grade data pipeline that pulls Google Workspace adoption metrics, aggregates them without storing any user-level PII, and loads them idempotently into BigQuery on a daily schedule.

You've learned how to:

  • Build a config-driven, multi-tenant pipeline that pulls Workspace metrics from the Admin SDK Reports API, aggregates per-user data to department-level segments, and keeps user identifiers in memory only.
  • Implement idempotent writes using partition decorators and WRITE_TRUNCATE so that re-runs and backfills never double-count or corrupt existing data.
  • Deploy to Cloud Run from source with an attached service account (no credential files shipped) and wire Cloud Scheduler for automated daily execution.
  • Secret Mission: Added a historical backfill mode with --start-date and --end-date CLI arguments and rate-limit-aware delays between API calls.

Ready to quiz yourself?