NeuralBlitz Google BigQuery
What I do
I am Google's serverless, highly scalable, multi-cloud data warehouse. I separate storage from compute, enabling instant elasticity and cost efficiency for analytical workloads. I support standard SQL with extensions for arrays, structs, and nested data. I provide automatic high availability, built-in ML capabilities, and seamless integration with the Google Cloud ecosystem. I am designed for petabyte-scale analytics with zero infrastructure management.
When to use me
- Large-scale data warehousing and business intelligence
- Log analytics and clickstream analysis
- Financial data analysis and fraud detection
- Ad tech and marketing analytics
- IoT data ingestion and analysis
- Machine learning feature stores
- Real-time streaming analytics
- Multi-source data unification
- Geospatial analysis with BigQuery GIS
- Cost-effective analytical queries on massive datasets
Core Concepts
- Storage-Compute Separation: Storage billed separately from query processing; independent scaling
- Slot-based Execution: Virtual CPUs (slots) process queries in parallel across distributed infrastructure
- Columnar Storage: Data stored in Capacitor columnar format for efficient analytical scans
- Partitioned Tables: Divide large tables by date, integer range, or ingestion time for cost optimization
- Clustered Tables: Organize data within partitions by specific columns for better query performance
- Streaming Inserts: Real-time data ingestion with immediate queryability
- BigQuery ML (BQML): Build and deploy ML models using SQL directly in BigQuery
- Streaming Analytics: Process real-time data streams with windowed aggregations
- Data Transfer Services: Automated data ingestion from SaaS applications and cloud storage
- Access Controls: IAM-based fine-grained permissions at project, dataset, and table levels
Code Examples
Basic Connection and Query Execution
from google.cloud import bigquery
import pandas as pd
client = bigquery.Client()
def execute_query(query, params=None):
job_config = bigquery.QueryJobConfig()
if params:
job_config.query_parameters = params
query_job = client.query(query, job_config=job_config)
return query_job.result()
def fetch_as_dataframe(query, params=None):
return execute_query(query, params).to_dataframe()
def get_user_summary():
return fetch_as_dataframe("""
SELECT
user_id,
user_email,
DATE(first_seen) as signup_date,
COUNT(*) as total_orders,
SUM(amount) as lifetime_value,
AVG(amount) as avg_order_value
FROM `analytics.orders`
GROUP BY user_id, user_email, signup_date
ORDER BY lifetime_value DESC
LIMIT 100
""")
def get_daily_metrics():
return fetch_as_dataframe("""
SELECT
DATE(created_at) as metric_date,
COUNT(DISTINCT user_id) as daily_active_users,
COUNT(*) as daily_orders,
SUM(amount) as daily_revenue,
AVG(amount) as avg_order_size
FROM `analytics.orders`
WHERE created_at >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 30 DAY)
GROUP BY DATE(created_at)
ORDER BY metric_date
""")
def search_products(search_term):
return fetch_as_dataframe("""
SELECT
product_id,
name,
category,
price,
brand
FROM `analytics.products`
WHERE LOWER(name) LIKE LOWER(@search_term)
ORDER BY popularity_score DESC
LIMIT 20
""", [bigquery.ScalarQueryParameter("search_term", "STRING", f"%{search_term}%")])
def get_recent_orders(limit=100):
return fetch_as_dataframe("""
SELECT
order_id,
user_id,
amount,
status,
created_at
FROM `analytics.orders`
ORDER BY created_at DESC
LIMIT @limit
""", [bigquery.ScalarQueryParameter("limit", "INT64", limit)])
Working with Nested and Repeated Data
import json
def load_nested_json_events(events_data):
rows = []
for event in events_data:
rows.append({
"event_id": event["event_id"],
"event_type": event["event_type"],
"user_id": event["user_id"],
"event_timestamp": event["created_at"],
"event_data": json.dumps(event.get("data", {})),
"user_agent": event.get("user_agent", ""),
"ip_address": event.get("ip", "")
})
errors = client.insert_rows_json("analytics.events", rows)
return len(errors) == 0
def query_user_behavior(user_id):
return fetch_as_dataframe("""
SELECT
event_type,
COUNT(*) as event_count,
MIN(event_timestamp) as first_seen,
MAX(event_timestamp) as last_seen
FROM `analytics.events`,
UNNEST(event_data) as ed
WHERE user_id = @user_id
GROUP BY event_type
ORDER BY event_count DESC
""", [bigquery.ScalarQueryParameter("user_id", "STRING", user_id)])
def analyze_user_sessions():
return fetch_as_dataframe("""
SELECT
user_id,
session_id,
MIN(event_timestamp) as session_start,
MAX(event_timestamp) as session_end,
TIMESTAMP_DIFF(MAX(event_timestamp), MIN(event_timestamp), SECOND) as duration_sec,
COUNT(*) as event_count
FROM (
SELECT
user_id,
event_id,
event_timestamp,
FIRST_VALUE(event_id) OVER (
PARTITION BY user_id
ORDER BY event_timestamp
ROWS BETWEEN 1 PRECEDING AND 1 PRECEDING
) as prev_event,
SUM(CASE
WHEN TIMESTAMP_DIFF(event_timestamp,
LAG(event_timestamp) OVER (PARTITION BY user_id ORDER BY event_timestamp),
MINUTE) > 30 THEN 1
ELSE 0
END) + 1 as session_id
FROM `analytics.events`
)
GROUP BY user_id, session_id
HAVING event_count > 1
ORDER BY duration_sec DESC
LIMIT 100
""")
def extract_nested_order_data():
return fetch_as_dataframe("""
SELECT
order_id,
customer.name as customer_name,
customer.email as customer_email,
(SELECT COUNT(*) FROM UNNEST(items)) as item_count,
(SELECT SUM(CAST(item.quantity AS INT64)) FROM UNNEST(items) item) as total_quantity,
(SELECT SUM(CAST(item.price AS FLOAT64) * CAST(item.quantity AS INT64))
FROM UNNEST(items) item) as total_amount
FROM `analytics.orders`
WHERE order_date >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 7 DAY)
""")
def analyze_struct_data():
return fetch_as_dataframe("""
SELECT
location.city,
location.country,
COUNT(*) as visit_count,
AVG(session.duration_seconds) as avg_session_duration
FROM `analytics.user_sessions`
WHERE session.duration_seconds > 0
GROUP BY location.city, location.country
ORDER BY visit_count DESC
""")
Streaming and Real-Time Analytics
def stream_order_event(order_data):
row = {
"order_id": order_data["order_id"],
"user_id": order_data["user_id"],
"amount": order_data["total"],
"items": json.dumps(order_data.get("items", [])),
"status": order_data.get("status", "pending"),
"created_at": order_data["created_at"]
}
errors = client.insert_rows_json("analytics.orders_stream", [row])
return len(errors) == 0
def get_realtime_dashboard():
return fetch_as_dataframe("""
SELECT
FORMAT_TIMESTAMP("%H:%M:%S", created_at, "UTC") as time_bucket,
COUNT(*) as orders_last_minute,
SUM(amount) as revenue_last_minute,
COUNT(DISTINCT user_id) as unique_users
FROM `analytics.orders_stream`
WHERE created_at >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 5 MINUTE)
GROUP BY time_bucket
ORDER BY time_bucket
""")
def detect_anomalies():
return fetch_as_dataframe("""
WITH hourly_stats AS (
SELECT
DATE(created_at) as date,
EXTRACT(HOUR FROM created_at) as hour,
COUNT(*) as order_count,
AVG(amount) as avg_amount,
STDDEV(amount) as std_amount
FROM `analytics.orders`
GROUP BY date, hour
),
recent_data AS (
SELECT
DATE(created_at) as date,
EXTRACT(HOUR FROM created_at) as hour,
COUNT(*) as order_count,
AVG(amount) as avg_amount
FROM `analytics.orders_stream`
WHERE created_at >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 1 HOUR)
GROUP BY date, hour
)
SELECT
r.date,
r.hour,
r.order_count,
r.avg_amount,
h.order_count as expected_orders,
h.avg_amount as expected_avg,
CASE
WHEN r.order_count > h.order_count * 2 THEN 'HIGH_VOLUME_ANOMALY'
WHEN r.avg_amount > h.avg_amount + 3 * h.std_amount THEN 'PRICE_ANOMALY'
ELSE 'NORMAL'
END as anomaly_type
FROM recent_data r
JOIN hourly_stats h ON r.date = h.date AND r.hour = h.hour
""")
def continuous_query():
return client.query("""
SELECT
event_type,
event_count,
event_timestamp
FROM `analytics.events`
WHERE event_timestamp > TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 10 MINUTE)
ORDER BY event_timestamp DESC
""")
Partitioning and Clustering
def create_partitioned_table():
return client.query("""
CREATE TABLE IF NOT EXISTS `analytics.orders_partitioned`
PARTITION BY DATE(created_at)
CLUSTER BY user_id, status
AS
SELECT * FROM `analytics.orders`
""")
def query_partitioned_table(start_date, end_date):
return fetch_as_dataframe("""
SELECT
DATE(created_at) as date,
COUNT(*) as order_count,
SUM(amount) as revenue
FROM `analytics.orders_partitioned`
WHERE created_at BETWEEN @start_date AND @end_date
GROUP BY DATE(created_at)
ORDER BY date
""", [
bigquery.ScalarQueryParameter("start_date", "TIMESTAMP", start_date),
bigquery.ScalarQueryParameter("end_date", "TIMESTAMP", end_date)
])
def get_table_info(dataset_id, table_id):
table_ref = client.dataset(dataset_id).table(table_id)
table = client.get_table(table_ref)
return {
"name": table.table_id,
"num_rows": table.num_rows,
"size_bytes": table.size_bytes,
"partitioned": table.partitioned,
"clustered_fields": [f.name for f in table.clustering_fields] if table.clustering_fields else [],
"schema": [(f.name, f.field_type) for f in table.schema]
}
def optimize_query_with_partitioning():
return fetch_as_dataframe("""
SELECT
user_id,
COUNT(*) as order_count
FROM `analytics.orders_partitioned`
WHERE created_at >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 90 DAY)
GROUP BY user_id
HAVING COUNT(*) > 10
ORDER BY order_count DESC
""")
def estimate_query_cost(query, terabytes_scanned=1.0):
job_config = bigquery.QueryJobConfig(dry_run=True)
query_job = client.query(query, job_config=job_config)
return {
"estimated_bytes": query_job.total_bytes_processed,
"estimated_cost_usd": query_job.total_bytes_processed / (1024**4) * 5.0
}
Machine Learning with BigQuery ML
def create_ml_model():
return client.query("""
CREATE OR REPLACE MODEL `analytics.user_churn_model`
OPTIONS(
model_type='logistic_reg',
input_label_cols=['churned'],
max_iterations=100
) AS
SELECT
user_id,
CASE WHEN DATEDIFF(MAX(order_date), CURRENT_DATE()) > 90 THEN 1 ELSE 0 END as churned,
COUNT(*) as total_orders,
SUM(amount) as total_spent,
AVG(amount) as avg_order_value,
COUNT(DISTINCT DATE(order_date)) as active_days
FROM `analytics.orders`
GROUP BY user_id
""")
def predict_churn():
return fetch_as_dataframe("""
SELECT
user_id,
predicted_churned as is_likely_to_churn,
prediction_probability
FROM ML.PREDICT(
MODEL `analytics.user_churn_model`,
(
SELECT
user_id,
COUNT(*) as total_orders,
SUM(amount) as total_spent,
AVG(amount) as avg_order_value,
COUNT(DISTINCT DATE(order_date)) as active_days
FROM `analytics.orders`
WHERE order_date >= TIMESTAMP_SUB(CURRENT_DATE(), INTERVAL 30 DAY)
GROUP BY user_id
)
)
ORDER BY prediction_probability DESC
LIMIT 100
""")
def train_recommendation_model():
return client.query("""
CREATE OR REPLACE MODEL `analytics.product_recommender`
OPTIONS(model_type='matrix_factorization') AS
SELECT
user_id,
product_id,
quantity as rating
FROM `analytics.order_items`
WHERE quantity > 0
""")
def get_recommendations(user_id, limit=10):
return fetch_as_dataframe("""
SELECT
product_id,
predicted_rating
FROM ML.RECOMMEND(
MODEL `analytics.product_recommender`,
(
SELECT @user_id as user_id,
ARRAY_AGG(DISTINCT product_id) as products
FROM `analytics.order_items`
WHERE user_id = @user_id
)
)
ORDER BY predicted_rating DESC
LIMIT @limit
""", [
bigquery.ScalarQueryParameter("user_id", "STRING", user_id),
bigquery.ScalarQueryParameter("limit", "INT64", limit)
])
def evaluate_model():
return fetch_as_dataframe("""
SELECT
* FROM ML.EVALUATE(
MODEL `analytics.user_churn_model`,
(
SELECT
user_id,
CASE WHEN DATEDIFF(MAX(order_date), CURRENT_DATE()) > 90 THEN 1 ELSE 0 END as churned,
COUNT(*) as total_orders,
SUM(amount) as total_spent
FROM `analytics.orders`
GROUP BY user_id
)
)
""")
Best Practices
- Use Partitioning: Partition large tables by date to reduce query costs and improve performance
- Cluster Your Data: Cluster partitioned tables on frequently filtered columns
- **Avoid SELECT ***: Only select needed columns to reduce data scanned and costs
- Use Streaming Inserts Judiciously: Streaming has higher cost than batch loading
- Leverage Cached Results: Query results are cached for 24 hours; reuse when possible
- Estimate Costs: Use dry_run to estimate query costs before execution
- Use Standard SQL: Legacy SQL is deprecated; use standard SQL with proper parameterization
- Optimize Joins: Filter before joining; use smaller table as driving table
- Monitor Slot Usage: Track slot utilization and adjust capacity as needed
- Use BQML for Feature Engineering: Build ML models directly in BigQuery for efficiency