2026-09-17, Analytics · Engineering · SERP API

Loading SERP and AI answer data into BigQuery: schema, loading and queries

The simplest reliable way to get SERP and AI answer results into BigQuery is to write each result your webhook receives as one line of newline-delimited JSON, batch-load those files hourly or daily into a date-partitioned table, and keep the full result in a JSON column next to a few typed fields and a nested array of organic results. Batch loading is free on BigQuery’s shared slot pool, the table stays cheap to query with partition filters, and you can extract new fields later without collecting again.

All BigQuery facts below are from Google’s documentation, checked 2026-09-17, and linked where used.

The BigQuery facts that shape the design

Consequences:

• Batch, don’t stream, unless you need minute-level freshness. A load per hour per table is 24 jobs a day against a limit of 1,500.
• Partition by day. 10,000 daily partitions is over 27 years of history.
• Keep the raw result as JSON. New questions become new SQL over history, not new collection.

The data you are loading

A finished async task arrives at your webhook as the same body GET /v1/async/task/{taskId} returns:

• task: id, taskType (GOOGLE, GOOGLE_NEWS, CHATGPT, PERPLEXITY, GEMINI, COPILOT, GROK, AIMODE), status, priority, createdAt, latencyMs, idempotencyKey.
• credits: creditsToCharge and creditsCharged.
• response: the engine’s result (success, result) when COMPLETED, or the error when FAILED.

The webhook body does not repeat the request payload, so keep your own record of what each task asked for, keyed by idempotencyKey or task id, and join it in when writing rows. Webhook receiver design covers storing deliveries durably first.

Result shapes differ by engine. Google Search returns organicResults[], peopleAlsoAsk[], relatedSearches[], ads[], localResults[], knowledgeGraph, shoppingCards[] and aioverview. Google News returns newsResults[]. AI engines return text and sources[], plus engine-specific fields (ChatGPT alone has entities[]). The schema below keeps what is common typed and the rest in JSON.

The schema

Two tables cover most programs: one row per observation, and your task plan.

create schema if not exists serp;

create table serp.observations (
  task_id          string    not null,
  task_type        string    not null,   -- GOOGLE, GOOGLE_NEWS, CHATGPT, ...
  status           string    not null,   -- COMPLETED or FAILED
  observed_at      timestamp not null,   -- when your receiver got it
  run_id           string,
  idempotency_key  string,
  query            string,               -- query or prompt from your plan
  country          string,
  hl               string,
  location         string,
  state            string,
  device           string,
  credits_charged  int64,
  latency_ms       int64,
  has_aioverview   bool,
  organic array<struct<
    position int64, page int64, title string, link string, domain string, snippet string
  >>,
  sources array<struct<
    position int64, url string, domain string, label string
  >>,
  result           json                  -- full engine result
)
partition by date(observed_at)
cluster by task_type, country, query
options (require_partition_filter = true);

Design notes:

• organic and sources as nested repeated fields. Position queries use unnest without parsing JSON on every read, and they stay columnar.
• domain precomputed. Nearly every rank or citation query groups by domain.
• result as JSON. Everything else (People Also Ask, AI Overview citations, shopping cards, map entries, search queries) stays queryable with JSON functions.
• require_partition_filter. A query without a date filter fails instead of scanning all history.
• Clustering on the columns most filters use; order matters, most selective filters that are always present first.

Writing rows from webhook deliveries

This Python function turns one delivery plus your stored plan entry into a row:

import json
from datetime import datetime, timezone
from urllib.parse import urlparse

def domain(url: str) -> str:
    host = urlparse(url or "").netloc.lower()
    return host[4:] if host.startswith("www.") else host

def to_row(delivery: dict, plan: dict) -> dict:
    task = delivery["task"]
    result = (delivery.get("response") or {}).get("result") or {}
    completed = task["status"] == "COMPLETED"
    organic = result.get("organicResults") or result.get("newsResults") or []
    sources = result.get("sources") or []
    return {
        "task_id": task["id"],
        "task_type": task["taskType"],
        "status": task["status"],
        "observed_at": datetime.now(timezone.utc).isoformat(),
        "run_id": plan.get("run_id"),
        "idempotency_key": task.get("idempotencyKey"),
        "query": plan["payload"].get("query") or plan["payload"].get("prompt"),
        "country": plan["payload"].get("country") or (plan["payload"].get("gl") or "").upper() or None,
        "hl": plan["payload"].get("hl"),
        "location": plan["payload"].get("location"),
        "state": plan["payload"].get("state"),
        "device": plan["payload"].get("device"),
        "credits_charged": delivery["credits"].get("creditsCharged"),
        "latency_ms": task.get("latencyMs"),
        "has_aioverview": bool(result.get("aioverview")) if task["taskType"] == "GOOGLE" else None,
        "organic": [
            {"position": r.get("position"), "page": r.get("page"), "title": r.get("title"),
             "link": r.get("link"), "domain": domain(r.get("link")), "snippet": r.get("snippet")}
            for r in organic
        ] if completed else [],
        "sources": [
            {"position": s.get("position"), "url": s.get("url"), "domain": domain(s.get("url")), "label": s.get("label")}
            for s in sources
        ] if completed else [],
        "result": result if completed else delivery.get("response"),
    }

def append_ndjson(path: str, row: dict) -> None:
    with open(path, "a", encoding="utf-8") as f:
        f.write(json.dumps(row, ensure_ascii=False) + "\n")

organic takes Google News newsResults too, since they share position, title, link and snippet. For AI engines, sources carries citations. Adjust per engine using the OpenAPI reference; field names differ between engines.

Write rows to an hourly file, such as observations/2026-09-17T14.ndjson, then upload to Cloud Storage and load.

Batch loading

With the bq CLI

gsutil cp observations/2026-09-17T14.ndjson gs://my-serp-bucket/observations/
bq load \
  --source_format=NEWLINE_DELIMITED_JSON \
  serp.observations \
  gs://my-serp-bucket/observations/2026-09-17T14.ndjson

The table already exists with its schema, so the load appends using it. Mistyped values fail the job; failed jobs count toward the 1,500 per table per day, so validate files before loading in a tight retry loop.

With the Python client

from google.cloud import bigquery

client = bigquery.Client()
job_config = bigquery.LoadJobConfig(
    source_format=bigquery.SourceFormat.NEWLINE_DELIMITED_JSON,
    write_disposition=bigquery.WriteDisposition.WRITE_APPEND,
)
job = client.load_table_from_uri(
    "gs://my-serp-bucket/observations/2026-09-17T*.ndjson",
    "my-project.serp.observations",
    job_config=job_config,
)
job.result()
print(job.output_rows, "rows loaded")

A wildcard URI loads a day’s hourly files in one job. One job per day per table is the cheapest cadence in quota terms; one per hour keeps data fresher and still uses under 2% of the per-table limit.

When to stream

Use the Storage Write API if a dashboard or alert must see results within minutes of the webhook. Its price is $0.025 per GiB after the first 2 TiB a month. For most rank tracking and AI visibility programs, daily reporting makes hourly batch loads the better trade.

Deduplication

Webhook deliveries can repeat, and a reconciliation pass can re-ingest a task you already have. Two options:

Read-time deduplication with a view:

create or replace view serp.observations_latest as
select *
from serp.observations
where date(observed_at) >= date_sub(current_date(), interval 400 day)
qualify row_number() over (partition by task_id order by observed_at desc) = 1;

Write-time deduplication by loading into a staging table and merging:

merge serp.observations t
using (
  select * from serp.observations_staging
  qualify row_number() over (partition by task_id order by observed_at desc) = 1
) s
on t.task_id = s.task_id and date(t.observed_at) >= date_sub(current_date(), interval 3 day)
when not matched then insert row;

The merge is a query and is billed for bytes scanned; the date condition on the target limits that to recent partitions. Read-time deduplication costs nothing extra at load time and is usually enough.

Example queries

All queries filter on the partition column, which require_partition_filter enforces.

Rank history for a domain

select date(observed_at) as day, query, country, device,
       min(o.position) as best_position
from serp.observations_latest, unnest(organic) as o
where date(observed_at) between '2026-08-01' and '2026-09-17'
  and task_type = 'GOOGLE'
  and o.domain = 'example.com'
group by day, query, country, device
order by query, country, device, day;

Queries where the domain is absent produce no row for that day; left-join against the observation list if you need explicit gaps.

Share of the top 10 by domain

select country, o.domain,
       count(*) as top10_appearances,
       round(count(*) / sum(count(*)) over (partition by country), 4) as share
from serp.observations_latest, unnest(organic) as o
where date(observed_at) = '2026-09-17'
  and task_type = 'GOOGLE'
  and o.position <= 10
group by country, o.domain
qualify row_number() over (partition by country order by count(*) desc) <= 20
order by country, top10_appearances desc;

AI Overview presence rate by market

select country, date(observed_at) as day,
       countif(has_aioverview) / count(*) as aio_rate,
       count(*) as observations
from serp.observations_latest
where date(observed_at) >= date_sub(current_date(), interval 28 day)
  and task_type = 'GOOGLE'
  and status = 'COMPLETED'
group by country, day
order by country, day;

Only meaningful if tasks requested include.aioverview; the field is null in results when not requested or not available.

Domains cited in AI Overviews

select net.reg_domain(json_value(src, '$.url')) as cited_domain, count(*) as citations
from serp.observations_latest,
     unnest(json_query_array(result, '$.aioverview.sources')) as src
where date(observed_at) >= date_sub(current_date(), interval 7 day)
  and task_type = 'GOOGLE'
group by cited_domain
order by citations desc
limit 50;

Citation rate in AI answers by engine

select task_type, country,
       countif(exists(select 1 from unnest(sources) s where s.domain = 'example.com')) / count(*) as citation_rate,
       countif(regexp_contains(lower(json_value(result, '$.text')), r'\bexample\b')) / count(*) as mention_rate,
       count(*) as answers
from serp.observations_latest
where date(observed_at) >= date_sub(current_date(), interval 28 day)
  and task_type in ('CHATGPT', 'PERPLEXITY', 'GEMINI', 'COPILOT', 'GROK')
  and status = 'COMPLETED'
group by task_type, country
order by task_type, citation_rate desc;

Mentions vs citations explains why both rates belong in the report.

Most frequent People Also Ask questions

select json_value(q, '$.question') as question, count(distinct query) as keywords
from serp.observations_latest,
     unnest(json_query_array(result, '$.peopleAlsoAsk')) as q
where date(observed_at) >= date_sub(current_date(), interval 30 day)
  and task_type = 'GOOGLE'
group by question
order by keywords desc
limit 100;

Collection health and cost

select date(observed_at) as day, task_type,
       countif(status = 'COMPLETED') as completed,
       countif(status = 'FAILED') as failed,
       sum(credits_charged) as credits,
       approx_quantiles(latency_ms, 100)[offset(50)] as p50_latency_ms
from serp.observations_latest
where date(observed_at) >= date_sub(current_date(), interval 14 day)
group by day, task_type
order by day desc, task_type;

Failed tasks are not charged, so credits should come only from completed rows. Compare completed with the number of tasks you planned per day to get completeness; SERP API reliability at scale describes that ledger.

Estimating BigQuery cost

1. Loading: free with batch loads.
2. Storage: measure average row bytes after a first load (INFORMATION_SCHEMA.TABLE_STORAGE or the table details page), multiply by rows per month, and apply the storage price on the pricing page; the first 10 GiB a month is free.
3. Queries: on-demand queries are billed per bytes scanned, $6.25 per TiB after the first free TiB each month. Dashboards that filter to recent partitions and clustered columns scan a small fraction of the table. Check a query’s estimate with a dry run (bq query --dry_run) before scheduling it.

Collection cost is separate and in credits: a Google Search async task is 3 credits, Google News 3, ChatGPT 5, Copilot 5, Grok 5, Gemini 4, Perplexity 4, AI Mode 4, with add-ons on some engines. See /pricing for plans.

Pitfalls

• Streaming by default. It costs money and adds nothing for daily reports.
• Retrying failed loads in a loop. Failed jobs count toward the per-table daily quota; fix the file first.
• No partition filter. Enforce it on the table.
• Only typed columns. Without the JSON column, every new question needs re-collection.
• Only JSON. Parsing JSON for every rank query is slower and scans more; keep the hot fields typed.
• Losing the request. The delivery body does not include the payload; join your plan before writing rows.
• Counting duplicates. Deduplicate on task_id.
• Mixing engines in one metric. Positions in organic results and citation positions in AI answers are different measures; keep task_type in every group by.

For collecting the data in the first place, see rank tracking with async batches and the async tasks docs.

Keep reading