Gemini API × BigQuery: AI-Powered Data Analysis — From

You've got terabytes of data sitting in BigQuery, but not everyone on your team can write SQL to extract the insights they need. What if anyone could simply ask a question in plain English and get an analyzed result back?

By combining Gemini API with Google BigQuery, you can build a natural language data analysis pipeline that generates SQL from questions, executes queries, and summarizes results — all automatically. The walkthrough below is a complete Python implementation—real code you can point at your own datasets.

What you'll learn:

How to architect a Gemini × BigQuery integration
Auto-generating SQL from natural language using schema-aware prompts
Summarizing query results into business-friendly insights
Production considerations for security and cost management

Who this is for: Developers and data analysts familiar with BigQuery basics who want to enhance their analysis workflow with Gemini AI.

Prerequisites and Setup

What you'll need

Google Cloud project — with BigQuery API and Vertex AI API (or a Google AI Studio API key) enabled
Python 3.10+ — using the google-cloud-bigquery and google-genai packages
A BigQuery dataset — we'll use the public bigquery-public-data.google_analytics_sample dataset as our example

Install the packages

pip install google-cloud-bigquery google-genai db-dtypes pandas

Configure your API key

import os
 
# Using Google AI Studio API key
os.environ["GOOGLE_API_KEY"] = "your-api-key-here"
 
# For Vertex AI, use service account authentication instead
# gcloud auth application-default login

Architecture Overview

The Gemini × BigQuery data analysis pipeline follows this flow:

User question (natural language)
    ↓
[Gemini API] Generate SQL using table schema context
    ↓
[BigQuery] Execute SQL and retrieve results
    ↓
[Gemini API] Summarize results and surface insights
    ↓
Output to report or dashboard

The key ingredient is passing table schema information to Gemini. With schema awareness, the model generates significantly more accurate SQL — correct column names, proper types, and valid joins.

Step 1: Extract BigQuery Schema Information

First, fetch the schema from your target tables and format it as context for Gemini.

from google.cloud import bigquery
 
client = bigquery.Client()
 
def get_table_schema(dataset_id: str, table_id: str) -> str:
    """Return table schema in a human-readable format."""
    table_ref = f"{client.project}.{dataset_id}.{table_id}"
    table = client.get_table(table_ref)
 
    schema_lines = [f"Table: {table_ref}"]
    schema_lines.append(f"Row count: ~{table.num_rows:,}")
    schema_lines.append(f"Columns:")
 
    for field in table.schema:
        nullable = "NULLABLE" if field.mode == "NULLABLE" else "REQUIRED"
        schema_lines.append(
            f"  - {field.name} ({field.field_type}, {nullable}): {field.description or 'No description'}"
        )
 
    return "\n".join(schema_lines)
 
# Example usage
schema_info = get_table_schema(
    "google_analytics_sample",
    "ga_sessions_20170801"
)
print(schema_info)

Sample output:

Table: your-project.google_analytics_sample.ga_sessions_20170801
Row count: ~170,366
Columns:
  - visitorId (INTEGER, NULLABLE): No description
  - visitNumber (INTEGER, NULLABLE): No description
  - visitId (INTEGER, NULLABLE): No description
  - visitStartTime (INTEGER, NULLABLE): No description
  ...

Step 2: Generate SQL from Natural Language with Gemini

Embed the schema into a system prompt and convert user questions into SQL queries.

from google import genai
 
# Initialize the Gemini client
genai_client = genai.Client()
 
def generate_sql(question: str, schema: str) -> str:
    """Generate a BigQuery SQL query from a natural language question."""
 
    system_prompt = f"""You are a BigQuery SQL expert.
Generate a SQL query that answers the user's question based on the following table schema.
 
{schema}
 
Rules:
- Output valid BigQuery Standard SQL only
- Return only the SQL query, no explanations
- Include appropriate LIMIT clauses to avoid returning excessive data
- Include the full project name in table references
- Do not include comments
"""
 
    response = genai_client.models.generate_content(
        model="gemini-2.5-flash",
        contents=question,
        config=genai.types.GenerateContentConfig(
            system_instruction=system_prompt,
            temperature=0.1,  # Low temperature for accurate SQL generation
        ),
    )
 
    # Strip markdown code block markers
    sql = response.text.strip()
    sql = sql.replace("```sql", "").replace("```", "").strip()
 
    return sql
 
# Example usage
question = "What are the top 10 traffic channels by session count?"
sql = generate_sql(question, schema_info)
print(sql)

Generated SQL example:

SELECT
  channelGrouping,
  COUNT(*) AS session_count
FROM
  `bigquery-public-data.google_analytics_sample.ga_sessions_20170801`
GROUP BY
  channelGrouping
ORDER BY
  session_count DESC
LIMIT 10

Validate SQL Before Execution

Never execute AI-generated SQL without validation. Use BigQuery's dry run feature to check syntax and estimate costs.

from google.cloud.bigquery import QueryJobConfig
 
def validate_sql(sql: str) -> dict:
    """Validate SQL with a dry run and return estimated bytes processed."""
    job_config = QueryJobConfig(dry_run=True, use_query_cache=False)
 
    try:
        query_job = client.query(sql, job_config=job_config)
        bytes_processed = query_job.total_bytes_processed
        return {
            "valid": True,
            "estimated_bytes": bytes_processed,
            "estimated_gb": round(bytes_processed / (1024**3), 4),
        }
    except Exception as e:
        return {"valid": False, "error": str(e)}
 
# Run validation
result = validate_sql(sql)
print(result)
# Output: {'valid': True, 'estimated_bytes': 148567232, 'estimated_gb': 0.1384}

Step 3: Execute Queries and Summarize Results with Gemini

Once validation passes, run the query and feed the results back to Gemini for analysis.

import pandas as pd
 
def execute_and_summarize(sql: str, question: str) -> str:
    """Execute SQL and summarize the results using Gemini."""
 
    # Execute the query
    df = client.query(sql).to_dataframe()
 
    # Convert the dataframe to a text representation
    result_text = df.to_string(index=False, max_rows=50)
 
    # Generate a summary with Gemini
    summary_prompt = f"""Summarize the following data analysis results in a way that a business stakeholder can understand.
 
## Original Question
{question}
 
## SQL Executed
{sql}
 
## Result Data
{result_text}
 
Please provide:
1. Key findings (the most important data points)
2. Notable trends or patterns
3. Recommended next actions
"""
 
    response = genai_client.models.generate_content(
        model="gemini-2.5-flash",
        contents=summary_prompt,
    )
 
    return response.text
 
# Example
summary = execute_and_summarize(sql, question)
print(summary)

Sample output:

## Analysis Summary

### Key Findings
- Organic Search dominates with approximately 43% of all sessions
- Social (22%) and Direct (18%) round out the top three channels

### Notable Patterns
- Paid Search accounts for less than 5% of sessions, highlighting the
  strength of organic acquisition strategies
- Referral traffic at ~10% suggests established external linking

### Recommended Actions
- Investigate Paid Search ROI to determine if increased spend is justified
- Break down Social traffic by platform for targeted optimization

Step 4: Build a Conversational Data Analysis Agent

Now let's combine all the pieces into an interactive analysis agent.

class GeminiBigQueryAgent:
    """A conversational agent for natural language BigQuery analysis."""
 
    def __init__(self, dataset_id: str, table_ids: list[str]):
        self.bq_client = bigquery.Client()
        self.genai_client = genai.Client()
        self.schemas = self._load_schemas(dataset_id, table_ids)
        self.history = []
 
    def _load_schemas(self, dataset_id, table_ids) -> str:
        """Load schemas for multiple tables."""
        all_schemas = []
        for tid in table_ids:
            schema = get_table_schema(dataset_id, tid)
            all_schemas.append(schema)
        return "\n\n".join(all_schemas)
 
    def ask(self, question: str) -> dict:
        """Process a question and return analysis results."""
        # 1. Generate SQL
        sql = generate_sql(question, self.schemas)
 
        # 2. Validate
        validation = validate_sql(sql)
        if not validation["valid"]:
            return {
                "error": f"SQL validation failed: {validation['error']}",
                "sql": sql,
            }
 
        # 3. Cost guard (require confirmation for queries over 1GB)
        if validation["estimated_gb"] > 1.0:
            return {
                "warning": f"Estimated scan: {validation['estimated_gb']}GB. Proceed?",
                "sql": sql,
            }
 
        # 4. Execute and summarize
        summary = execute_and_summarize(sql, question)
 
        # 5. Add to history
        self.history.append({
            "question": question,
            "sql": sql,
            "bytes": validation["estimated_bytes"],
        })
 
        return {
            "question": question,
            "sql": sql,
            "summary": summary,
            "estimated_cost_gb": validation["estimated_gb"],
        }
 
# Usage
agent = GeminiBigQueryAgent(
    dataset_id="google_analytics_sample",
    table_ids=["ga_sessions_20170801"]
)
 
result = agent.ask("What's the conversion rate by device category?")
print(result["summary"])

Production Security and Cost Management

SQL Injection Protection

Since AI-generated SQL is executed directly, implement guardrails to ensure only read operations are permitted.

import re
 
FORBIDDEN_PATTERNS = [
    r"\bDROP\b",
    r"\bDELETE\b",
    r"\bUPDATE\b",
    r"\bINSERT\b",
    r"\bCREATE\b",
    r"\bALTER\b",
    r"\bTRUNCATE\b",
]
 
def is_safe_sql(sql: str) -> bool:
    """Allow only read-only SQL statements."""
    upper_sql = sql.upper()
    for pattern in FORBIDDEN_PATTERNS:
        if re.search(pattern, upper_sql):
            return False
    return upper_sql.strip().startswith("SELECT") or upper_sql.strip().startswith("WITH")

Cost Management Best Practices

BigQuery charges based on the volume of data scanned, so cost control is essential.

Always dry run first: Use validate_sql to check estimated costs before execution
Set byte limits: Use QueryJobConfig(maximum_bytes_billed=1_000_000_000) for a 1GB cap
Leverage partitioning: Date-partitioned tables dramatically reduce scan volume
Use caching: Identical queries are served from BigQuery's automatic cache at no charge

# Execute with a cost cap
job_config = QueryJobConfig(
    maximum_bytes_billed=1_000_000_000  # 1GB limit
)
df = client.query(sql, job_config=job_config).to_dataframe()

Summary

Combining Gemini API with BigQuery opens the door to natural language data analysis that anyone on your team can use — no SQL expertise required. The architecture we've covered — schema-aware SQL generation, dry-run validation, and AI-powered summarization — provides a solid foundation for building internal analytics tools, Slack bots, or custom dashboards.

Start by experimenting with public datasets to get comfortable with the workflow, then gradually introduce your own data. The patterns we've covered here scale well from prototype to production.

If you're interested in building more sophisticated data analysis agents, check out [Build an AI Data Analysis Agent with Gemini API]((/articles/gemini-api/gemini-api-ai-data-analysis-agent) for in-depth agent design patterns. For improving SQL generation accuracy, [Getting Started with Natural Language SQL Conversion in Gemini]((/articles/gemini-api/gemini-text-to-sql-guide) is another great resource.

To dive deeper into the topics covered in this article,

Gemini API × BigQuery: AI-Powered Data Analysis — From Natural Language Queries to Actionable Insights