Estimating per-query cost

If you are on an on-demand pricing plan, you can obtain a cost estimate for a query before submitting it to the service. The BigQuery web user interface (UI) validates queries and provides an estimate of the amount of data that will be processed. You can see the number of bytes the query will process before you run the query by clicking the Query Validator. If you are using the bq command-line client, specify --dry_run to take a look at the query plan and the amount of data processed before invoking the query for real. Dry runs are free. Knowing the amount of data that will be processed by the query, you can use the Google Cloud Platform (GCP) Pricing Calculator to estimate the cost of the query in dollars. Tools such as BigQuery Mate and superQuery provide a price estimate directly but might not have access to information about negotiated discounts. As of this writing, the cost is five dollars per terabyte (for US and EU multiregions) after the free tier usage of one free terabyte per month is exceeded. Note that BigQuery needs to read only the columns that are referenced in your queries, and partitioning and clustering can further reduce the amount of data that needs to be scanned (and hence the cost).

When invoking a query, you can specify the --maximum_bytes_billed parameter to put a limit on the amount of data that a query can process. If the bytes scanned in the query exceeds the maximum bytes that can be billed, the query will fail without incurring a charge. You can also manage costs by requesting a custom quota from the GCP Cloud Console for a limit on the amount of query data processed per day. You can set this limit at a per-project or per-user level.

Finding the most expensive queries

When trying to control costs, it can be helpful to create a short list of queries to focus on. You can do this by querying the INFORMATION_SCHEMA associated with a project to find the most expensive queries:

SELECT
   job_id
   , query
   , user_email
   , total_bytes_processed
   , total_slot_ms
FROM `some-project`.INFORMATION_SCHEMA.JOBS_BY_PROJECT
WHERE EXTRACT(YEAR FROM creation_time) = 2019
ORDER BY total_bytes_processed DESC
LIMIT 5

The preceding query lists the five most expensive queries in 2019 in some project based on total_bytes_processed. If you are on flat-rate pricing, you might choose to order the queries based on total_slot_ms instead.

Obtaining query plan information from the job details

The information listed about each stage of a completed query in the job information includes the timestamps at which the stage was started and finished, the total number of records read and written, and the number of bytes written across all workers in order to process this query. For example, try executing the following query:

SELECT
  start_station_name,
  AVG(duration) AS duration,
  COUNT(duration) AS num_trips
FROM
  `bigquery-public-data`.london_bicycles.cycle_hire
GROUP BY
  start_station_name
ORDER BY
  num_trips DESC
LIMIT
  5

Now you can list the job details by invoking the REST API:⁹

JOBID=8adbf3fd-e310-44bb-9c6e-88254958ccac   # CHANGE
access_token=$(gcloud auth application-default print-access-token)
PROJECT=$(gcloud config get-value project)
curl --silent 
     -H "Authorization: Bearer $access_token" 
     -X GET 
     "https://www.googleapis.com/bigquery/v2/projects/$PROJECT/jobs/$JOBID"

For example, the data about the first stage includes the following information about the time spent waiting for data, the ratio of I/O to compute in the query, and data read, shuffled, and written out:

"waitRatioAvg": 0.058558558558558557,
"readRatioAvg": 0.070270270270270274,
"computeRatioAvg": 0.86036036036036034
...
"shuffleOutputBytes": "356596",
"shuffleOutputBytesSpilled": "0",
"recordsRead": "24369201",
"recordsWritten": "6138",
"parallelInputs": "7",

Because this is primarily an input stage, we care primarily about read and shuffle performance—because this stage kept the processor busy 86% of the time and no data needed to be spilled to disk, it appears that bottlenecks (if any) in this query are not related to I/O. If this query is slow, we’ll need to look elsewhere, perhaps at reasons for the relatively low number of parallelization (7); in this case, this low count is fine because the input data of 24 million records is rather small and quite capable of being processed in seven chunks.

Visualizing the query plan information

Much of the information about the query plan in the job details is represented visually in the BigQuery web UI. Open the query and click the “Execution details” tab to see a depiction of the query plan. This includes the overall timing as well as a breakdown of this timing at the level of the steps that make up each stage of the query.

Figure 7-4. Key measures from the query execution details tab; performance optimization will typically focus on reducing the slot time and/or bytes shuffled

The overall timing depicted in Figure 7-5 indicates seven parallel inputs being active in the first 0.6 seconds, and then two more units being activated.

Figure 7-5. Timing data from the query plan information

Further details of the stages and steps are provided visually as well, as illustrated in Figure 7-6.

Figure 7-6. Stages and steps from the query plan information

From Figure 7-6, we see that the query has three stages. The first stage (S00) is the input stage, the second (S01) consists of sorting, and the third (S02) is the output stage. The timing of each stage is shown visually as well in Figure 7-6. The ratios that were available numerically in the JSON response to the job details request are depicted in the color bars shown in Figure 7-7.¹⁰

Figure 7-7. Stage timing from the query plan information

The dark colors indicate the average time spent waiting, reading, computing, or writing. Here, most of the time is spent performing computation, and a little bit of time is spent waiting. The wait stage is somewhat variable (the intermediate color indicates the maximum wait, and the difference between the average and the maximum is an indicator of variability)—here, the maximum wait time is nearly double the average wait time. Reading, on the other hand, is quite consistent, whereas the writing overhead is negligible.

Zooming in on the input stage, shown in Figure 7-8, we see that it consists of three steps: reading two columns (duration and start_station_name, referred to as $1 and $2) from the BigQuery table, aggregating, and writing the output.

Figure 7-8. The steps that form the first stage

The aggregation step consists of three operations: grouping by start_station_name (recall that $2 refers to this input column), finding the average duration ($1) within each shard, and maintaining the count of non-null durations. The writing stage writes the groups ($30), average duration ($20), and count of duration ($21) to an intermediate output, distributing the output by the hash of the station name. In case more than one worker is needed in the next stage, the hash of the station name controls which workers process which parts of the stage_00 output.

Looking back again at Figure 7-6, you can see that the first stage is carried out in parallel over seven workers, while the remaining two stages are carried out on a single worker. The input stage reads in 24.4 million rows and writes out around 6,140 rows, which totals 348 KB. These rows are sorted by the second stage, and five rows are written to the third stage. When we have a stage with only one worker, we should ensure that the memory load on that worker is well within the bounds of what a single worker can handle (we cover ways to do so later in this chapter)—348 KB definitely qualifies, and so this query should not pose any performance issues due to limits on the resources available to a single worker.

Another option to visualize the BigQuery query plan is to use the BigQuery Visualizer at https://bqvisualiser.appspot.com/, as shown in Figure 7-9.

Figure 7-9. Visualizing a BigQuery job using the BigQuery Visualizer

The visualizer becomes especially useful for complex queries with tens of stages, which can be difficult to comprehend from just the synopsis available in the BigQuery web UI.

Be purposeful in SELECT

Because BigQuery uses columnar file formats, the fewer the columns that are read in a SELECT, the less the amount of data that needs to be read. In particular, doing a SELECT * reads every column of every row in the table, making it quite slow and expensive. The exception is when you use a SELECT * in a subquery and then reference only a few fields in an outer query; the BigQuery optimizer will be smart enough to read only the columns that are absolutely required.

Explicitly list the columns that you want to see in the final result. For example, it is much more efficient to find the bike_id responsible for the longest duration trip in the dataset by doing the following:

SELECT
  bike_id
  , duration
FROM
  `bigquery-public-data`.london_bicycles.cycle_hire
ORDER BY duration DESC
LIMIT 1

A less efficient method is this:

SELECT
  *
FROM
  `bigquery-public-data`.london_bicycles.cycle_hire
ORDER BY duration DESC
LIMIT 1

The first query took us 1.8 seconds and cost 372 MB, whereas the second one took us 5.5 seconds (three times slower) and cost 2.59 GB (seven times costlier).

Reducing data being read

When tuning a query, it is important to start with the data that is being read and consider whether it is possible to reduce it. Suppose that you want to find the typical duration of the most common one-way rentals; you could do the following:

SELECT
  MIN(start_station_name) AS start_station_name
  , MIN(end_station_name) AS end_station_name
  , APPROX_QUANTILES(duration, 10)[OFFSET(5)] AS typical_duration
  , COUNT(duration) AS num_trips
FROM
  `bigquery-public-data`.london_bicycles.cycle_hire
WHERE
  start_station_id != end_station_id 
GROUP BY
  start_station_id, end_station_id
ORDER BY num_trips DESC
LIMIT 10

This takes 14.7 seconds when we run it, and it yields the following:

The details of the query indicate that the sorting (for the approximate quantiles for every station pair) requires a repartition of the outputs of the input stage, but most of the time is spent during computation, as demonstrated in Figure 7-10.

Figure 7-10. This query requires two repartition stages, but most of the time is spent in computation

Nevertheless, we can reduce the I/O overhead of the query if we do the filtering and grouping using the station name rather than the station ID, because we will need to read fewer columns:

SELECT
  start_station_name
  , end_station_name
  , APPROX_QUANTILES(duration, 10)[OFFSET(5)] AS typical_duration
  , COUNT(duration) AS num_trips
FROM
  `bigquery-public-data`.london_bicycles.cycle_hire
WHERE
  start_station_name != end_station_name 
GROUP BY
  start_station_name, end_station_name
ORDER BY num_trips DESC
LIMIT 10

This query avoids the need to read the two ID columns and finishes in 9.6 seconds, a 30% increase in speed. This increase is caused by the downstream effects of reading less data: the query requires one less repartition, and fewer workers (10, versus 19 earlier) for the sort, as shown in Figure 7-11.

Figure 7-11. By taking advantage of the 1:1 relationship between station_id and station_name, we are able to read fewer columns, remove one stage, and use fewer workers for the sort

The query result remains the same because there is a 1:1 relationship between the station name and the station ID.

Reducing the number of expensive computations

Suppose that you want to find the total distance traveled by each bicycle in the dataset. A naive way to do this would be to find the distance traveled in each trip undertaken by each bicycle and sum up the distances:

WITH trip_distance AS (
  SELECT
     bike_id
     , ST_Distance(ST_GeogPoint(s.longitude, s.latitude),
                   ST_GeogPoint(e.longitude, e.latitude)) AS distance
  FROM
     `bigquery-public-data`.london_bicycles.cycle_hire,
     `bigquery-public-data`.london_bicycles.cycle_stations s,
     `bigquery-public-data`.london_bicycles.cycle_stations e
  WHERE 
     start_station_id = s.id
     AND end_station_id = e.id
)
 
SELECT 
  bike_id
  , SUM(distance)/1000 AS total_distance
FROM trip_distance
GROUP BY bike_id
ORDER BY total_distance DESC
LIMIT 5

This query takes 7.1 seconds (44 seconds of slot time) and shuffles 1.69 MB. The result is that some bicycles have been ridden nearly 6,000 kilometers:

Computing the distance is a pretty expensive operation, and we can avoid joining the cycle_stations table against the cycle_hire table if we precompute the distances between all pairs of stations:

WITH stations AS (
  SELECT
     s.id AS start_id
     , e.id AS end_id
     , ST_Distance(ST_GeogPoint(s.longitude, s.latitude),
                   ST_GeogPoint(e.longitude, e.latitude)) AS distance
  FROM
     `bigquery-public-data`.london_bicycles.cycle_stations s,
     `bigquery-public-data`.london_bicycles.cycle_stations e
),

The rest of the query is quite similar, except that the join is against the table of precomputed distances:

trip_distance AS (
  SELECT
     bike_id
     , distance
  FROM
     `bigquery-public-data`.london_bicycles.cycle_hire,
     stations
  WHERE 
     start_station_id = start_id
     AND end_station_id = end_id
)
 
SELECT 
  bike_id
  , SUM(distance)/1000 AS total_distance
FROM trip_distance
GROUP BY bike_id
ORDER BY total_distance DESC
LIMIT 5

Now the query takes only 31.5 seconds of slot time (a 30% increase in speed) in spite of having to shuffle more data (33.44 MB) between nodes.

Caching intermediate results

It is possible to improve overall performance at the expense of increased I/O by taking advantage of temporary tables and materialized views. For example, suppose that you have a number of queries that start out by finding the typical duration of trips between a pair of stations:

WITH typical_trip AS (
SELECT
  start_station_name
  , end_station_name
  , APPROX_QUANTILES(duration, 10)[OFFSET(5)] AS typical_duration
  , COUNT(duration) AS num_trips
FROM
  `bigquery-public-data`.london_bicycles.cycle_hire
GROUP BY
  start_station_name, end_station_name
)

The WITH clause (also called a common table expression) improves readability but does not improve query speed or cost because results are not cached. The same applies to views and subqueries as well. If you find yourself using a WITH clause, a view, or a subquery often, one way to potentially improve performance is to store the result in a table (or materialized view):¹¹

CREATE OR REPLACE TABLE ch07eu.typical_trip AS
SELECT
  start_station_name
  , end_station_name
  , APPROX_QUANTILES(duration, 10)[OFFSET(5)] AS typical_duration
  , COUNT(duration) AS num_trips
FROM
  `bigquery-public-data`.london_bicycles.cycle_hire
GROUP BY
  start_station_name, end_station_name

Let’s use the WITH clause to find days when bicycle trips are much longer than usual:

SELECT 
   EXTRACT (DATE FROM start_date) AS trip_date
   , APPROX_QUANTILES(duration / typical_duration, 10)[OFFSET(5)] AS ratio
   , COUNT(*) AS num_trips_on_day
FROM 
  `bigquery-public-data`.london_bicycles.cycle_hire AS hire
JOIN typical_trip AS trip
ON 
   hire.start_station_name = trip.start_station_name 
   AND hire.end_station_name = trip.end_station_name
   AND num_trips > 10
GROUP BY trip_date
HAVING num_trips_on_day > 10
ORDER BY ratio DESC
LIMIT 10

This takes 19.1 seconds and processes 1.68 GB. Now, let’s use the table:

SELECT 
   EXTRACT (DATE FROM start_date) AS trip_date
   , APPROX_QUANTILES(duration / typical_duration, 10)[OFFSET(5)] AS ratio
   , COUNT(*) AS num_trips_on_day
FROM 
  `bigquery-public-data`.london_bicycles.cycle_hire AS hire
JOIN ch07eu.typical_trip AS trip
ON 
   hire.start_station_name = trip.start_station_name 
   AND hire.end_station_name = trip.end_station_name
   AND num_trips > 10
GROUP BY trip_date
HAVING num_trips_on_day > 10
ORDER BY ratio DESC
LIMIT 10

This takes 10.3 seconds (a 50% increase in speed because the computation is avoided) and processes 1.72 GB (a slight increase in cost because the new table is now being read). Both queries return the same result, that trips on Christmas take longer than usual:

The table ch07eu.typical_trip is not refreshed when new data is added to the cycle_hire table. One way to solve this problem of stale data is to use a materialized view or to schedule queries to update the table periodically. You should measure the cost of such updates to see whether the improvement in query performance makes up for the extra cost of keeping the intermediate table up to date.

Accelerating queries with BI Engine

If there are tables that you access frequently in Business Intelligence (BI) settings, such as dashboards with aggregations and filters, one way to speed up your queries is to employ BI Engine. It will automatically store relevant pieces of data in memory (either actual columns from the table or derived results) and will use a specialized query processor tuned for working with mostly in-memory data. You can use the BigQuery Admin Console to reserve the amount of memory (up to a current maximum of 10 GB) that BigQuery should use for its cache, as depicted in Figure 7-12.

Figure 7-12. Reserve memory for caching table data by setting up a BI Engine reservation

Make sure to reserve this memory in the same region as the dataset you are querying. Then BigQuery will start to cache tables, parts of tables, and aggregations in memory and serve results faster.

A primary use case for BI Engine is for tables that are accessed from dashboard tools such as Google Data Studio. By providing memory allocation for a BI Engine reservation, you can make dashboards that rely on a BigQuery backend much more responsive.

Denormalization

One way to improve the read performance and avoid joins is to give up on storing data efficiently and instead add redundant copies of data. This is called denormalization. Thus, instead of storing the bicycle station latitudes and longitudes separately from the cycle hire information, we could create a denormalized table:

CREATE OR REPLACE TABLE ch07eu.london_bicycles_denorm AS
SELECT
  start_station_id
  , s.latitude AS start_latitude
  , s.longitude AS start_longitude
  , end_station_id
  , e.latitude AS end_latitude
  , e.longitude AS end_longitude 
FROM
  `bigquery-public-data`.london_bicycles.cycle_hire as h
JOIN
  `bigquery-public-data`.london_bicycles.cycle_stations as s
ON
  h.start_station_id = s.id
JOIN
  `bigquery-public-data`.london_bicycles.cycle_stations as e
ON
  h.end_station_id = e.id

Then all subsequent queries will not need to carry out the join because the table will contain the necessary location information for all trips:

In this case, you are trading off storage and reading more data for the computational expense of a join. It is quite possible that the cost of reading more data from disk will outweigh the cost of the join—you should measure whether denormalization brings performance benefits.

Avoiding self-joins of large tables

Self-joins happen when a table is joined with itself. Although BigQuery supports self-joins, they can lead to performance degradation if the table being joined with itself is very large. In many cases, you can avoid the self-join by taking advantage of SQL features such as aggregation and window functions.

Let’s look at an example. One of the BigQuery public datasets is the dataset of baby names published by the US Social Security Administration. It is possible to query the dataset to find the most common male names for the year 2015 in the state of Massachusetts:

SELECT 
  name
  , number AS num_babies
FROM `bigquery-public-data`.usa_names.usa_1910_current
WHERE gender = 'M' AND year = 2015 AND state = 'MA'
ORDER BY num_babies DESC
LIMIT 5

Here’s the result of that query:

Similarly, the most common female names (gender = 'F') for the year 2015 in Massachusetts were as follows:

What are the most common names assigned to both male and female babies in the country over all the years in the dataset? A naive way to solve this problem involves reading the input table twice and doing a self-join:

WITH male_babies AS (
SELECT 
  name
  , number AS num_babies
FROM `bigquery-public-data`.usa_names.usa_1910_current
WHERE gender = 'M'
),
female_babies AS (
SELECT 
  name
  , number AS num_babies
FROM `bigquery-public-data`.usa_names.usa_1910_current
WHERE gender = 'F'
),
both_genders AS (
SELECT 
  name
  , SUM(m.num_babies) + SUM(f.num_babies) AS num_babies
  , SUM(m.num_babies) / (SUM(m.num_babies) + SUM(f.num_babies)) AS frac_male
FROM male_babies AS m
JOIN female_babies AS f
USING (name)
GROUP BY name
)
 
SELECT * FROM both_genders
WHERE frac_male BETWEEN 0.3 and 0.7
ORDER BY num_babies DESC
LIMIT 5

This took 74 seconds and yielded the following:

To add insult to injury, the answer is also wrong: as much as we like the name Jordan, the entire US population is only around 330 million, so there cannot have been 982 million babies with that name. The self-join unfortunately joins across state and year boundaries.¹²

A faster, more elegant (and correct!) solution is to recast the query to read the input only once and avoid the self-join completely. This took only 2.4 seconds—a 30-times increase in speed:

WITH all_babies AS (
SELECT 
  name 
  , SUM(IF(gender = 'M', number, 0)) AS male_babies
  , SUM(IF(gender = 'F', number, 0)) AS female_babies
FROM `bigquery-public-data.usa_names.usa_1910_current`
GROUP BY name
),
 
both_genders AS (
SELECT
  name
  , (male_babies + female_babies) AS num_babies
  , SAFE_DIVIDE(male_babies, male_babies + female_babies) AS frac_male
FROM all_babies
WHERE male_babies > 0 AND female_babies > 0
)
 
SELECT * FROM both_genders
WHERE frac_male BETWEEN 0.3 and 0.7
ORDER BY num_babies desc
limit 5

Here’s the result, in case you’re curious:

Reducing the data being joined

It is possible to carry out the previous query with an efficient join as long as we reduce the amount of data being joined by grouping the data by name and gender early on:

with all_names AS (
  SELECT name, gender, SUM(number) AS num_babies
  FROM `bigquery-public-data`.usa_names.usa_1910_current
  GROUP BY name, gender
),
 
male_names AS (
   SELECT name, num_babies
   FROM all_names
   WHERE gender = 'M'
),
 
female_names AS (
   SELECT name, num_babies
   FROM all_names
   WHERE gender = 'F'
),
 
ratio AS (
  SELECT 
    name
    , (f.num_babies + m.num_babies) AS num_babies
    , m.num_babies / (f.num_babies + m.num_babies) AS frac_male
  FROM male_names AS m
  JOIN female_names AS f
  USING (name)
)
 
SELECT * from ratio
WHERE frac_male BETWEEN 0.3 and 0.7
ORDER BY num_babies DESC
LIMIT 5

The early grouping serves to trim the data early in the query, before the query performs a JOIN. That way, shuffling and other complex operations execute only on the much smaller data and remain quite efficient. This query finished in two seconds and returned the correct result.

Using a window function instead of self-join

Suppose that you want to find the duration between a bike being dropped off and it being rented again; in other words, the duration that a bicycle stays at the station. This is an example of a dependent relationship between rows. It might appear that the only way to solve this is to join the table with itself, matching the end_date of one trip against the start_date of the next.

You can, however, avoid a self-join by using a window function (we cover window functions in Chapter 8):

SELECT
  bike_id
  , start_date
  , end_date
  , TIMESTAMP_DIFF(
       start_date, 
       LAG(end_date) OVER (PARTITION BY bike_id ORDER BY start_date),
       SECOND) AS time_at_station
FROM `bigquery-public-data`.london_bicycles.cycle_hire
LIMIT 5

Here’s the result of that query:

Notice that the first row has a null for time_at_station because we don’t have a timestamp for the previous dropoff. After that, the time_at_station tracks the difference between the previous dropoff and the current pickup.

Using this, we can compute the average time that a bicycle is unused at each station and rank stations by that measure:

WITH unused AS (
SELECT
  bike_id
  , start_station_name
  , start_date
  , end_date
  , TIMESTAMP_DIFF(start_date, LAG(end_date) OVER (PARTITION BY bike_id ORDER BY
start_date), SECOND) AS time_at_station
FROM `bigquery-public-data`.london_bicycles.cycle_hire
)
 
SELECT
  start_station_name
  , AVG(time_at_station) AS unused_seconds
FROM unused
GROUP BY start_station_name
ORDER BY unused_seconds ASC
LIMIT 5

From this query, we learn that bicycles turn over the fastest at the following stations:

Joining with precomputed values

Sometimes it can be helpful to precompute functions on smaller tables and then join with the precomputed values rather than repeat an expensive calculation each time.

For example, suppose that you want to find the pair of stations between which our customers ride bicycles at the fastest pace. To compute the pace¹³ (minutes per kilometer) at which they ride, we need to divide the duration of the ride by the distance between stations.

We could create a denormalized table with distances between stations and then compute the average pace:

with denormalized_table AS (
  SELECT
    start_station_name
    , end_station_name
    , ST_DISTANCE(ST_GeogPoint(s1.longitude, s1.latitude),
                  ST_GeogPoint(s2.longitude, s2.latitude)) AS distance
    , duration
 FROM
    `bigquery-public-data`.london_bicycles.cycle_hire AS h
 JOIN
     `bigquery-public-data`.london_bicycles.cycle_stations AS s1
 ON h.start_station_id = s1.id
  JOIN
     `bigquery-public-data`.london_bicycles.cycle_stations AS s2
 ON h.end_station_id = s2.id
),
 
durations AS (
  SELECT
    start_station_name
    , end_station_name
    , MIN(distance) AS distance
    , AVG(duration) AS duration
    , COUNT(*) AS num_rides
  FROM
     denormalized_table
  WHERE
     duration > 0 AND distance > 0
  GROUP BY start_station_name, end_station_name
  HAVING num_rides > 100
)
 
SELECT
    start_station_name
    , end_station_name
    , distance
    , duration
    , duration/distance AS pace
FROM durations
ORDER BY pace ASC
LIMIT 5

This query invokes the geospatial function ST_DISTANCE once for each row in the cycle_hire table (24 million times), takes 16.1 seconds, and processes 1.86 GB.

Alternatively, we can use the cycle_stations table to precompute the distance between every pair of stations (this is a self-join) and then join it with the reduced-size table of average duration between stations:

with distances AS (
  SELECT 
    a.id AS start_station_id
    , a.name AS start_station_name
    , b.id AS end_station_id
    , b.name AS end_station_name
    , ST_DISTANCE(ST_GeogPoint(a.longitude, a.latitude),
                ST_GeogPoint(b.longitude, b.latitude)) AS distance
 FROM
    `bigquery-public-data`.london_bicycles.cycle_stations a
 CROSS JOIN
     `bigquery-public-data`.london_bicycles.cycle_stations b
 WHERE a.id != b.id
),
 
durations AS (
  SELECT
    start_station_id
    , end_station_id
    , AVG(duration) AS duration
    , COUNT(*) AS num_rides
  FROM
    `bigquery-public-data`.london_bicycles.cycle_hire
  WHERE
    duration > 0
  GROUP BY start_station_id, end_station_id
  HAVING num_rides > 100
)
 
SELECT
    start_station_name
    , end_station_name
    , distance
    , duration
    , duration/distance AS pace
FROM distances
JOIN durations
USING (start_station_id, end_station_id)
ORDER BY pace ASC
LIMIT 5

The recast query with the more efficient joins takes only 5.4 seconds, an increase in speed of three times, and processes 554 MB, a reduction in cost of nearly four times.

JOIN versus denormalization

What if we were to store the distance traveled in each trip in a denormalized table?

CREATE OR REPLACE TABLE ch07eu.cycle_hire AS
SELECT
  start_station_name
  , end_station_name
  , ST_DISTANCE(ST_GeogPoint(s1.longitude, s1.latitude),
               ST_GeogPoint(s2.longitude, s2.latitude)) AS distance
  , duration
FROM
  `bigquery-public-data`.london_bicycles.cycle_hire AS h
JOIN
  `bigquery-public-data`.london_bicycles.cycle_stations AS s1
ON h.start_station_id = s1.id
JOIN
  `bigquery-public-data`.london_bicycles.cycle_stations AS s2
ON h.end_station_id = s2.id

Querying this table returns results in 8.7 seconds and processes 1.6 GB—in other words, it’s 60% slower and about three times more expensivethan the previous query. In this instance, therefore, joining with a smaller table turns out to be more efficient than querying a larger, denormalized table. However, this is the sort of thing that you need to measure for your particular use case. You will see later how you can efficiently store data at differing levels of granularity in a single denormalized table with nested and repeated fields.

Limiting large sorts

Suppose that you want to go through the bike rentals and number them 1, 2, 3, and so on, in the order that the rentals ended. We could do that by using the ROW_NUMBER() function (we cover window functions in Chapter 8):

SELECT 
  rental_id
  , ROW_NUMBER() OVER(ORDER BY end_date) AS rental_number
FROM `bigquery-public-data`.london_bicycles.cycle_hire
ORDER BY rental_number ASC
LIMIT 5

Here’s the result:

However, this query takes 29.9 seconds to process just 372 MB because it needs to sort the entirety of the london_bicycles dataset on a single worker. Had we processed a larger dataset, it would have overwhelmed that worker.

In such cases, we might want to consider whether it is possible to limit the large sorts and distribute them. Indeed, it is possible to extract the date from the rentals and then sort trips within each day:

WITH rentals_on_day AS (
SELECT 
  rental_id
  , end_date
  , EXTRACT(DATE FROM end_date) AS rental_date
FROM `bigquery-public-data.london_bicycles.cycle_hire`
)
 
SELECT 
  rental_id
  , rental_date
  , ROW_NUMBER() OVER(PARTITION BY rental_date ORDER BY end_date) AS
rental_number_on_day
FROM rentals_on_day
ORDER BY rental_date ASC, rental_number_on_day ASC
LIMIT 5

This takes 8.9 seconds (an increase in speed of three times) because the sorting can be done on just a single day of data at a time. It yields the rental number on a day-by-day basis:

Data skew

The same problem of overwhelming a worker (in this case, overwhelming the memory of the worker) can happen during an ARRAY_AGG with GROUP BY if one of the keys is much more common than the others.¹⁴

Because there are more than three million GitHub repositories and the commits are well distributed among them, this query succeeds:

SELECT 
  repo_name
  , ARRAY_AGG(STRUCT(author, committer, subject, message, trailer, difference,
encoding) ORDER BY author.date.seconds)
FROM `bigquery-public-data.github_repos.commits`, UNNEST(repo_name) AS repo_name
GROUP BY repo_name

However, most of the people using GitHub live in only a few time zones, so grouping by the time zone fails—we are asking a single worker to sort a significant fraction of 750 GB:

SELECT 
  author.tz_offset, ARRAY_AGG(STRUCT(author, committer, subject, message, 
trailer, difference, encoding) ORDER BY author.date.seconds)
FROM `bigquery-public-data.github_repos.commits`
GROUP BY author.tz_offset

One solution is to add a LIMIT to the ORDER BY:

SELECT 
  author.tz_offset, ARRAY_AGG(STRUCT(author, committer, subject, message, 
trailer, difference, encoding) ORDER BY author.date.seconds LIMIT 1000)
FROM `bigquery-public-data.github_repos.commits`
GROUP BY author.tz_offset

If you do require sorting all of the data, use more granular keys (i.e., distribute the group’s data over more workers) and then aggregate the results corresponding to the desired key. For example, instead of grouping only by the time zone, it is possible to group by time zone and repo_name and then aggregate across repositories to get the actual answer for each time zone:

SELECT 
  repo_name, author.tz_offset
  , ARRAY_AGG(STRUCT(author, committer, subject, message, trailer, difference,
encoding) ORDER BY author.date.seconds)
FROM `bigquery-public-data.github_repos.commits`, UNNEST(repo_name) AS repo_name
GROUP BY repo_name, author.tz_offset

Optimizing user-defined functions

Invoking a JavaScript user-defined function (UDF) requires a V8 subprocess to be launched, and this degrades performance. JavaScript UDFs are computationally expensive and have access to limited memory, so reducing the amount of data processed by the UDF can help improve performance.

Although BigQuery supports UDFs in JavaScript, opt to write your UDFs using SQL wherever possible; SQL is distributed and optimized by BigQuery natively. If you are writing a UDF in SQL, there is no performance difference between embedding the SQL function directly in the query or using temporary and permanent functions. The reason to use an SQL UDF is for reusability, composability, and readability.
 

Approximate count

For example, you can find the number of unique GitHub repositories by using:

SELECT 
  COUNT(DISTINCT repo_name) AS num_repos
FROM `bigquery-public-data`.github_repos.commits, UNNEST(repo_name) AS repo_name

This query takes 7.1 seconds to compute the correct result of 3,348,576. On the other hand, the following query takes 3.2 seconds (an increase in speed of two times) and returns an approximate result of 3,400,927, which overestimates the correct answer by 1.5%:

SELECT 
  APPROX_COUNT_DISTINCT(repo_name) AS num_repos
FROM `bigquery-public-data`.github_repos.commits, UNNEST(repo_name) AS repo_name

On smaller datasets, however, there might be no advantage. Let’s look at an example that finds the total number of unique bicycles in the london_bicycles dataset:

SELECT 
  COUNT(DISTINCT bike_id) AS num_bikes
FROM `bigquery-public-data`.london_bicycles.cycle_hire

This takes 0.9 seconds and returns the correct result of 13,705. Using the approximate counterpart takes 1.6 seconds (which is slower than the exact query) and returns an approximate result of 13,699:

SELECT 
  APPROX_COUNT_DISTINCT(bike_id) AS num_bikes
 FROM `bigquery-public-data`.london_bicycles.cycle_hire

Approximate top

Other available approximate functions include APPROX_QUANTILES to compute percentiles, APPROX_TOP_COUNT to find the top elements, and APPROX_TOP_SUM to compute top elements based on the sum of an element. 

Here’s an example of using APPROX_TOP_COUNT to find the five most frequently rented bicycles:

SELECT 
  APPROX_TOP_COUNT(bike_id, 5) AS num_bikes
FROM `bigquery-public-data`.london_bicycles.cycle_hire

This yields the following:

Note that the result is a single row and consists of an array of values so that ordering is preserved.

To find the top five stations based on duration of bicycle rentals, you can use APPROX_TOP_SUM:

SELECT 
  APPROX_TOP_SUM(start_station_name, duration, 5) AS num_bikes
FROM `bigquery-public-data`.london_bicycles.cycle_hire
WHERE duration > 0

Here is the result of that query:

HLL functions

In addition to the just-described APPROX_* functions (which carry out the entire approximate aggregation algorithm), BigQuery also supports the HyperLogLog++ (HLL++) algorithm, which allows you to break down the count-distinct problem into three separate operations:

1. Initialize a set, called an HLL sketch, by adding new elements to it by using HLL_COUNT.INIT

2. Find the cardinality (count) of an HLL sketch by using HLL_COUNT.EXTRACT

3. Merge two HLL sketches into a single sketch by using HLL_COUNT.MERGE_PARTIAL

In addition, HLL_COUNT.MERGE combines steps 2 and 3, computing the count from a set of HLL sketches.

For example, here’s a query that finds the count of distinct stations in the London bicycles table regardless of whether a trip started or ended at the station:

WITH sketch AS (
SELECT
    HLL_COUNT.INIT(start_station_name) AS hll_start
   , HLL_COUNT.INIT(end_station_name) AS hll_end
FROM `bigquery-public-data`.london_bicycles.cycle_hire
)
 
SELECT 
  HLL_COUNT.MERGE(hll_start) AS distinct_start
  , HLL_COUNT.MERGE(hll_end) AS distinct_end
  , HLL_COUNT.MERGE(hll_both) AS distinct_station
FROM sketch, UNNEST([hll_start, hll_end]) AS hll_both

This returns the following:

Of course, you also can achieve this by using APPROX_COUNT_DISTINCT directly:

SELECT
   APPROX_COUNT_DISTINCT(start_station_name) AS distinct_start
  , APPROX_COUNT_DISTINCT(end_station_name) AS distinct_end
  , APPROX_COUNT_DISTINCT(both_stations) AS distinct_station
FROM 
  `bigquery-public-data`.london_bicycles.cycle_hire
  , UNNEST([start_station_name, end_station_name]) AS both_stations

This yields the same result and is much simpler to read and understand. Mostly, therefore, you would use the APPROX_ variants.

One reason to use the HLL functions might be that you need to employ manual aggregation or prevent storage of specific columns. Suppose that your data has the schema user_id, date, product, country, and you need to compute the number of distinct users. However, the column user_id is personally identifying, and so you would prefer to not store it indefinitely. In such a case, you can compute a manual aggregation using HLL_COUNT.INIT, as follows:

INSERT INTO approx_distinct_users_agg AS 
SELECT date, product, country, HLL_COUNT.INIT(user_id) AS sketch
GROUP BY date, product, country, sketch

Now you don’t need to store user_id; you only need to store the sketch. Whenever you need to compute any higher-level aggregation, you can do the following:

SELECT date, HLL_COUNT.MERGE(sketch)
FROM approx_distinct_users_agg 
GROUP BY date

Compressed, partial responses

When invoking the REST API directly, you can minimize network overhead by accepting compressed, partial responses. To accept compressed responses, you can specify in the HTTP header that you will accept gzip and make sure that the string “gzip” appears in the name of the user-agent—for example:

Accept-Encoding: gzip
User-Agent: programName (gzip)

Then the responses are compressed using gzip.

By default, responses from BigQuery contain all of the fields promised in the documentation. However, if we know what part of the response we are interested in, we can ask BigQuery to send back only that bit of the response, thus lowering the network overhead. For example, earlier in this chapter, we looked at how to get back the complete job details using the Jobs API. If you are interested in only a subset of the full response (for example, only the steps in the query plan), you can specify the field(s) of interest to limit the size of the response:¹⁵

JOBSURL="https://www.googleapis.com/bigquery/v2/projects/$PROJECT/jobs"
FIELDS="statistics(query(queryPlan(steps)))"
curl --silent 
    -H "Authorization: Bearer $access_token" 
    -H "Accept-Encoding: gzip" 
    -H "User-Agent: get_job_details (gzip)" 
    -X GET 
    "${JOBSURL}/${JOBID}?fields=${FIELDS}" 
| zcat

Note that we are also specifying that we accept gzip encoding.

Batching multiple requests

When using the REST API, it is possible to batch multiple BigQuery API calls by using the multipart/mixed content type and nesting HTTP requests in each of the parts. The body of each part specifies the HTTP operation (GET, PUT, etc.), path portion of the URL, headers, and body. The server’s response is a single HTTP response with a multipart/mixed content type, with the parts being responses (in order) to the requests that form the batched request. Even though the responses are in order, the server might execute the calls in any order. Thus you should treat the batched request as a parallel execution.

Here’s an example of sending a batched request to get some query plan details of the last five queries in our project. You first use the BigQuery command-line tool to get the five most recent successful jobs:¹⁶

# The 5 most recent successful jobs
JOBS=$(bq ls -j -n 50 | grep SUCCESS | head -5 | awk '{print $1}')

The request goes to the batch endpoint for BigQuery:

BATCHURL="https://www.googleapis.com/batch/bigquery/v2"
JOBSPATH="/projects/$PROJECT/jobs"
FIELDS="statistics(query(queryPlan(steps)))"

Using the URL path, you can form the individual requests:

request=""
for JOBID in $JOBS; do
read -d '' part << EOF
 
--batch_part_starts_here
GET ${JOBSPATH}/${JOBID}?fields=${FIELDS}
 
EOF
request=$(echo "$request"; echo "$part")
done

Then you can send the request to the batch endpoint as a multipart request:

curl --silent 
    -H "Authorization: Bearer $access_token" 
    -H "Content-Type: multipart/mixed; boundary=batch_part_starts_here" 
    -X POST 
    -d "$request" 
    "${BATCHURL}"

Bulk reads using BigQuery Storage API

In Chapter 5, we discussed using the BigQuery REST API and associated client libraries to list table data and get query results. The REST API provides the data in record-oriented, paginated views that are more conducive to relatively small result sets. Yet, with the advent of machine learning and distributed Extract, Transform, and Load (ETL) tools, external tools now require fast, efficient bulk access to BigQuery’s managed storage. Such bulk read access is provided by the BigQuery Storage API via Remote Procedure Call (RPC)–based protocol. With the BigQuery Storage API, structured data is sent over the wire in a binary serialization format that maps more closely to the columnar format in which the data is stored. This allows for additional parallelism among multiple consumers for a set of results.

It is unlikely that you will use the BigQuery Storage API directly¹⁷ if you are an end user. Instead, you will take advantage of tools such as Cloud Dataflow, Cloud Dataproc, TensorFlow, AutoML, and others that employ the Storage API to read directly from BigQuery’s managed storage instead of going through the BigQuery API. Because the Storage API directly accesses stored data, permission to access the BigQuery Storage API is distinct from the existing BigQuery API. The BigQuery Storage API must be enabled independently of enabling BigQuery.

The BigQuery Storage API provides several benefits to tools that read directly from BigQuery’s managed storage. For example, consumers can read disjoint a set of rows from a table using multiple streams (enabling distributed reads from different workers in Cloud Dataproc, for example), dynamically shard these streams (thus reducing tail latency, which can be a significant problem for MapReduce jobs), select a subset of columns to read (enabling machine learning frameworks to read only the features used by the model), filter column values (reducing the data transmitted over the network), and still ensure snapshot consistency (i.e., read data as of a specific point in time).

In Chapter 5, we looked at using the Jupyter Magics %%bigquery to load the result of queries into pandas DataFrames. However, those datasets were relatively small—on the order of a dozen to a few hundred rows. What if you want to load the entire london_bicycles dataset (24 million rows) into a pandas DataFrame? In that case, it is possible to specify an option on the Magics to load the data into the pandas DataFrame using the Storage API rather than the BigQuery API. You will need to first install the Storage API Python client library with Avro and pandas support. You can do this within Jupyter by using the following:

%pip install google-cloud-bigquery-storage[fastavro,pandas]

Then use the %%bigquery Magics as before, but add the Storage API option:

%%bigquery df --use_bqstorage_api --project $PROJECT
SELECT 
  start_station_name 
  , end_station_name
  , start_date
  , duration
FROM `bigquery-public-data`.london_bicycles.cycle_hire

Note that we are taking advantage of the ability of the Storage API to provide direct access to individual columns; it is not necessary to read the entire BigQuery table into the pandas DataFrame. If it so happens that the amount of data returned by the query is comfortably small, the Magics falls back automatically to the BigQuery API. Therefore, there is no harm in always using this flag in your notebook cells. To turn --use_bqstorage_api on by default in all the Magics cells within a notebook session, you can set a context flag:

import google.cloud.bigquery.magics
google.cloud.bigquery.magics.context.use_bqstorage_api = True

Internal versus external data sources

Even though BigQuery supports querying directly from external sources such as Google Cloud Storage, Cloud Bigtable, and Google Sheets, the fastest query performance will be obtained if you use native tables.

We recommend that you use BigQuery as your analytics data warehouse for all your structured and semi-structured data. Use external data sources for staging (Google Cloud Storage), real-time ingest (Cloud Pub/Sub, Cloud Bigtable), or transactional updates (Cloud SQL, Cloud Spanner). Then, as described in Chapter 4, set up a periodic data pipeline to load the data from these external sources into BigQuery.

If your use case is such that you need to query the data from Google Cloud Storage, store it in a compressed, columnar format (e.g., Parquet) if you can. Use row-based formats such as JSON or comma-separated values (CSV) only as a last resort.

Setting up life cycle management on staging buckets

If you are loading data into BigQuery by staging it in Google Cloud Storage first, consider deleting the Google Cloud Storage data after the data is loaded. If you are performing ETL on the data to load it into BigQuery (so that the data in BigQuery is heavily transformed or is only a subset), you might want to retain the raw data in Google Cloud Storage. In such cases, reduce costs by creating life cycle rules on buckets, to downgrade the Google Cloud Storage storage class.

To enable life cycle management on a bucket so that data in multiregional or standard classes older than 30 days is moved to Nearline Storage, and Nearline data older than 90 days is moved to Coldline Storage:

gsutil lifecycle set lifecycle.yaml gs://some_bucket/

In this example, the file lifecycle.yaml contains this content:

{
"lifecycle": {
  "rule": [
  {
     "action": {
       "type": "SetStorageClass",
       "storageClass": "NEARLINE"
     },
     "condition": {
       "age": 30,
       "matchesStorageClass": ["MULTI_REGIONAL", "STANDARD"]
     }
  },
  {
     "action": {
       "type": "SetStorageClass",
       "storageClass": "COLDLINE"
     },
     "condition": {
       "age": 90,
       "matchesStorageClass": ["NEARLINE"]
     }
  }
]}}

You can use life cycle management not just to change an object’s class but also to delete objects older than a certain threshold.¹⁸

Storing data as arrays of structs

One of the BigQuery public datasets is a dataset of cyclonic storms (hurricanes, typhoons, cyclones, etc.) as observed and measured by various meteorological agencies around the world. Cyclonic storms can last up to a few weeks, and observations are carried out once every three hours or so. Suppose that you want to query this dataset to find all of the storms in 2018, the maximum wind speed attained by each storm over its lifetime, and the time and location of the storm when this maximum wind speed was reached. This query pulls out the necessary information from the public dataset:

SELECT
  sid, number, basin, name,
  ARRAY_AGG(STRUCT(iso_time, usa_latitude, usa_longitude, usa_wind) ORDER BY
usa_wind DESC LIMIT 1)[OFFSET(0)].*
FROM
  `bigquery-public-data`.noaa_hurricanes.hurricanes
WHERE
  season = '2018'
GROUP BY
  sid, number, basin, name
ORDER BY number ASC

We are extracting the storm ID (sid), the storm number within the season, the basin, and the name of the storm (if named) and then finding the array of observations made of this storm, ranking the observations in descending order of wind speed, and taking the highest wind speed observed for each storm. The storms themselves are ordered by number. The result consists of 88 rows, and looks something like this:

The query processed 41.7 MB and took 1.4 seconds. The first row is of a storm named Bolaven that reached a maximum wind speed of 29 kph on January 2, 2018, at 18:00 UTC.

Because observations are carried out by multiple meteorological agencies, it is possible to standardize this data using nested fields and store the structs in BigQuery, as follows:¹⁹

CREATE OR REPLACE TABLE ch07.hurricanes_nested AS
 
SELECT sid, season, number, basin, name, iso_time, nature, usa_sshs,
       STRUCT(usa_latitude AS latitude, usa_longitude AS longitude, usa_wind AS
wind, usa_pressure AS pressure) AS usa,
       STRUCT(tokyo_latitude AS latitude, tokyo_longitude AS longitude,
           tokyo_wind AS wind, tokyo_pressure AS pressure) AS tokyo,
       ... AS cma,
       ... AS hko,
       ... AS newdelhi,
       ... AS reunion,
       ... bom,
       ... AS wellington,
       ... nadi
FROM `bigquery-public-data`.noaa_hurricanes.hurricanes

Querying this table is similar to querying the original table, except that field names change a bit (usa.latitude instead of usa_latitude):

SELECT
  sid, number, basin, name,
  ARRAY_AGG(STRUCT(iso_time, usa.latitude, usa.longitude, usa.wind) ORDER BY
usa.wind DESC LIMIT 1)[OFFSET(0)].*
FROM
  ch07.hurricanes_nested
WHERE
  season = '2018'
GROUP BY
  sid, number, basin, name
ORDER BY number ASC

The query processes the same amount of data and takes the same time as the original query on the public dataset. Using nested fields (i.e., structs) does not change the query speed or cost, although it might make your query more readable.

Because there are multiple observations of the same storm over its life cycle, we can change the storage to have a single row per storm and store an array of observations for each storm:

CREATE OR REPLACE TABLE ch07.hurricanes_nested_track AS
 
SELECT sid, season, number, basin, name,
 ARRAY_AGG(
   STRUCT(
       iso_time,
       nature,
       usa_sshs,
       STRUCT(usa_latitude AS latitude, usa_longitude AS longitude, usa_wind AS
wind, usa_pressure AS pressure) AS usa,
       STRUCT(tokyo_latitude AS latitude, tokyo_longitude AS longitude,
           tokyo_wind AS wind, tokyo_pressure AS pressure) AS tokyo,
       ... AS cma,
       ... AS hko,
       ... AS newdelhi,
       ... AS reunion,
       ... bom,
       ... AS wellington,
       ... nadi
   ) ORDER BY iso_time ASC ) AS obs
FROM `bigquery-public-data`.noaa_hurricanes.hurricanes
GROUP BY sid, season, number, basin, name

Notice that we are now storing sid, season, and so on as scalar columns because they do not change over the lifetime of the storm. The remaining data that changes for each observation is stored as an array of structs. Here’s how querying this table now looks:²⁰

SELECT
  number, name, basin, 
  (SELECT AS STRUCT iso_time, usa.latitude, usa.longitude, usa.wind 
          FROM UNNEST(obs) ORDER BY usa.wind DESC LIMIT 1).*
FROM ch07.hurricanes_nested_track
WHERE season = '2018'
ORDER BY number ASC

The result is the same, but this time the query processes only 14.7 MB (a reduction in cost of three times) and finishes in one second (a 30% improvement in speed). Why does this improvement in performance happen? When we store the data as an array, the number of rows in the table reduces dramatically (from 682,000 to 14,000),²¹ because there is now only one row per storm instead of one row per observation time. Then when we filter the rows by checking for season, BigQuery is able to discard many related observations simultaneously, as shown in Figure 7-13.

Figure 7-13. Nested and repeated fields can speed up query performance by allowing BigQuery to discard many related observations simultaneously

Another benefit is that we no longer need to duplicate rows of data when we have unequal levels of granularity within the same table. You can store both the granular-level individual hurricane latitude and longitude data as well as the high-level hurricane ID, name, and season data in the same table. And because BigQuery stores table data as highly compressed individual columns, you can query and process the high-level data without the cost of operating over the rows of granular data—it is now stored as an array of values per hurricane.

For example, if you simply want to query the number of storms by year, you could query just the columns you want from the details:

WITH hurricane_detail AS (
 
SELECT sid, season, number, basin, name,
 ARRAY_AGG(
   STRUCT(
       iso_time,
       nature,
       usa_sshs,
       STRUCT(usa_latitude AS latitude, usa_longitude AS longitude, usa_wind AS
wind, usa_pressure AS pressure) AS usa,
       STRUCT(tokyo_latitude AS latitude, tokyo_longitude AS longitude, tokyo_wind
AS wind, tokyo_pressure AS pressure) AS tokyo
   ) ORDER BY iso_time ASC ) AS obs
FROM `bigquery-public-data`.noaa_hurricanes.hurricanes
GROUP BY sid, season, number, basin, name
)
 
SELECT 
  COUNT(sid) AS count_of_storms,
  season
FROM hurricane_detail
GROUP BY season 
ORDER BY season DESC

The preceding query processes 27 MB instead of the 56 MB it would have had to process had we not used nested, repeated fields.

Nested fields by themselves do not improve performance, although they can improve readability by essentially prejoining other related tables into a single location. Nested, repeated fields, on the other hand, are extremely advantageous from a performance standpoint. Consider using nested, repeated fields in your schema, because they have the potential to provide you a significant boost in speed and lower query costs whenever you have queries that filter on a column that is not nested or repeated (season, in our case).

A key drawback to nested, repeated fields is that you cannot easily stream to this table if the streaming updates involve adding elements to existing arrays—it is no longer as simple as appending a row to the table: you now need to mutate the existing row. Of course, because the hurricane data is updated with new observations, this drawback would be quite significant, and it explains why the public dataset of hurricanes does not use nested, repeated fields.

On the other hand, the public dataset of GitHub commits (bigquery-public-data.github_repos.commits) uses a nested, repeated field (repo_name) to store the list of repositories affected by a commit. This doesn’t change over time. So using a nested, repeated field for repo_name provides a speedup to queries that filter on any other field.

Storing data as geography types

Within the BigQuery public dataset of utility data is a table of polygon boundaries of US zip codes (available in bigquery-public-data.utility_us.zipcode_area) and another table of polygon boundaries of US cities (bigquery-public-data.utility_us.us_cities_area). The US zip code geometry column (zipcode_geom) is a string,²⁴ whereas the city geometry column (city_geom) is a geography type.

From these two tables, it is possible to get a list of all the zip codes for Santa Fe, New Mexico:²⁵

SELECT name, zipcode 
FROM `bigquery-public-data`.utility_us.zipcode_area
JOIN `bigquery-public-data`.utility_us.us_cities_area
ON ST_INTERSECTS(ST_GeogFromText(zipcode_geom), city_geom)
WHERE name LIKE '%Santa Fe%'

The query took 51.9 seconds and processed 305.5 MB to produce the following:

Why did the query take so long? This is not because ST_INTERSECTS is expensive; it is mainly because the ST_GeogFromText function needs to compute the S2 cells²⁶ and build a GEOGRAPHY type corresponding to each zip code.

We can change the zip code table to do this computation beforehand and store the geometry as a GEOGRAPHY type:

CREATE OR REPLACE TABLE ch07.zipcode_area AS
SELECT 
  * REPLACE(ST_GeogFromText(zipcode_geom) AS zipcode_geom)
FROM
  `bigquery-public-data`.utility_us.zipcode_area

The new dataset is 131.8 MB, which is somewhat larger than the 116.5 MB of the original table. However, the trade-off is that queries against the table can take advantage of S2 coverings and become much faster. Thus, the following query finishes in 5.3 seconds (a speed increase of 10 times) and processes 320.8 MB (a slight increase in cost if you are using on-demand pricing):

SELECT name, zipcode 
FROM ch07.zipcode_area
JOIN `bigquery-public-data`.utility_us.us_cities_area
ON ST_INTERSECTS(zipcode_geom, city_geom)
WHERE name LIKE '%Santa Fe%'

The performance advantages of storing geographic data as GEOGRAPHY types instead of as strings or primitives are very compelling. This is why the utility_us dataset is deprecated (it’s still publicly accessible so as to not break already written queries). We recommend that you use the table bigquery-public-data.geo_us_boundaries.us_​zip_codes, which uses GEOGRAPHY types and is kept up to date.

Antipattern: Table suffixes and wildcards

If filtering by year is very common, one way to cut down on the data being read is to store the data in multiple tables, with the name of each table suffixed by the year. This way, querying the data for 2015 does not require traversing rows corresponding to all the years but can instead just read the cycle_hire_2015 table.

Let’s go ahead and create such a table using the following:

CREATE OR REPLACE TABLE ch07eu.cycle_hire_2015 AS (
  SELECT * FROM `bigquery-public-data`.london_bicycles.cycle_hire
  WHERE EXTRACT(YEAR from start_date) = 2015
)

Now let’s create it with a year-sharded table. The query finishes in one second (a speed increase of three times) and needs to process only 345 MB (a cost saving of three times):

SELECT 
  start_station_name 
  , AVG(duration) AS avg_duration
 FROM ch07eu.cycle_hire_2015
 GROUP BY start_station_name
 ORDER BY avg_duration DESC
 LIMIT 5

It is possible to use wildcards and table suffixes to search for multiple years:

SELECT 
  start_station_name 
  , AVG(duration) AS avg_duration
FROM `ch07eu.cycle_hire_*`
WHERE _TABLE_SUFFIX BETWEEN '2015' AND '2016'
GROUP BY start_station_name
ORDER BY avg_duration DESC
LIMIT 5

Partitioned tables

Partitioned tables allow you to store all of your related data in a single logical table but also efficiently query a subset of that data. If, for example, you store the last year’s worth of data but usually query only the last week, partitioning by time allows you to run queries that will need to scan only the last seven days’ worth of partitions. This can save orders of magnitude in query cost, slot utilization, and time.

The year-named tables discussed in the previous section are inefficient: BigQuery needs to maintain a copy of the schema and metadata for each of the sharded tables and verify permissions on each of the queried tables. Also, a single query cannot query more than 1,000 tables, and this might make querying the entire dataset difficult if your tables are sharded by date rather than by year (1,000 tables is not even three years of data if the tables are date-sharded). Furthermore, streaming into date-sharded tables can lead to the need for clock and time zone synchronization among multiple clients. The recommended best practice, therefore, is to use partitioned tables.

A partitioned table is a special table that is divided into partitions, with the partitions managed by BigQuery. We can create a partitioned version of the London cycle_hire dataset using the following:

CREATE OR REPLACE TABLE ch07eu.cycle_hire_partitioned
   PARTITION BY DATE(start_date) AS
SELECT * FROM `bigquery-public-data`.london_bicycles.cycle_hire

CREATE OR REPLACE TABLE ch07eu.cycle_hire_partitioned
   PARTITION BY DATE(start_date) 
   OPTIONS(partition_expiration_days=1000,    
           require_partition_filter=true) AS
SELECT * FROM `bigquery-public-
data`.london_bicycles.cycle_hire

ALTER TABLE ch07eu.cycle_hire_partitioned
SET OPTIONS(require_partition_filter=true)

Then, to find the stations with the longest average rentals in 2015, query the partitioned table, making sure to use the partition column (start_date) in the filter clause:

SELECT 
  start_station_name 
  , AVG(duration) AS avg_duration
FROM ch07eu.cycle_hire_partitioned
WHERE start_date BETWEEN '2015-01-01' AND '2015-12-31'
GROUP BY start_station_name
ORDER BY avg_duration DESC
LIMIT 5

The query takes one second and processes only 419.4 MB, a little more than the year-sharded table (because of the need to read the start_date column), but it is still a saving over having to read the full dataset. Note, however, that there are disadvantages to formulating the query as follows:

SELECT 
  start_station_name 
  , AVG(duration) AS avg_duration
FROM ch07eu.cycle_hire_partitioned
WHERE EXTRACT(YEAR FROM start_date) = 2015
GROUP BY start_station_name
ORDER BY avg_duration DESC
LIMIT 5

This will end up processing 1 GB and will not yield any savings on the amount of data processed. To obtain the benefits of partitioning, the BigQuery runtime must be able to statically determine the partition filters.

Clustering by the partitioning column

Partitioned tables are partitioned by date (whether it is a column or it is by ingestion time). If you need hour-level partitioning, one option is to use date-based partitioning and then cluster by hour along with whatever other attributes are appropriate.

So a common pattern is to cluster by the same column as you partition by. For example, if you partition by event_time, a timestamp at which your log events occur, this will let you do very fast and efficient queries over arbitrary time periods that are smaller than the day boundary used in the partitioning. You could, for instance, query over only the last 10 minutes of data, and you wouldn’t need to scan anything older than that.

In a partitioned table, each partition consists of a single day of data, and BigQuery maintains the metadata necessary to ensure that queries, load jobs, and Data Definition Language (DDL)/DML statements all take advantage and maintain the integrity of the partitions.

Reclustering

In a clustered table, BigQuery sorts the data based on the values in the clustering columns and organizes them into storage blocks that are optimally sized for efficient scanning and discarding of unnecessary data. However, unlike with partitioning, BigQuery does not maintain the sorting of data within clusters as data is streamed to it. BigQuery will recluster the data periodically in order to maintain efficient data pruning and scan speed. You can see how efficiently clustered the table is by looking at the clustering_ratio of the table (1.0 is completely optimal).

Updating a table using DML forces a recluster of the partition being updated. For example, assume that you periodically receive a table of corrections (which consist of cycle hires that were somehow missed in the previous updates). Using a MERGE statement such as the following will cause a recluster of any partitions that are updated:

MERGE ch07eu.cycle_hire_clustered all_hires
USING ch07eu.cycle_hire_corrections some_month
ON all_hires.start_station_name = some_month.start_station_name
WHEN MATCHED 
  AND all_hires._PARTITIONTIME = DATE(some_month.start_date) THEN
  INSERT (rental_id, duration, ...)
  VALUES (rental_id, duration, ...)

UPDATE ch07eu.cycle_hire_clustered 
SET start_station_id = 300
WHERE start_station_id = 300
AND start_date > TIMESTAMP_SUB(CURRENT_TIMESTAMP(),
INTERVAL 1 DAY)

Table 7-1 summarizes the differences between partitioning and clustering and might help you choose between partitioning a table based on a column and clustering the table by that column.

Side benefits of clustering

Remember that we mentioned that running SELECT * ... LIMIT 10 in BigQuery is an antipattern because it ends up billing you for the full scan of the table. For clustered tables, this is not true. When you’re reading from a clustered table, BigQuery will pass along any optimizations that can be done to prevent reading data. So if you do SELECT * ... LIMIT 10 on a clustered table, the execution engine will be able to stop reading data as soon as 10 rows have been returned. Because of how the query engine employs a number of parallel workers, any of which could happen to finish first, the amount of data that is scanned is not deterministic. On the plus side, you will end up with queries that cost you much less on large tables.

A surprising side effect of the “early stop” cost reduction is that you can get performance benefits even if you don’t filter on the clustering columns—if you filter by columns that are correlated to your clustering columns, BigQuery might be able to read less data!

Suppose that you have a table with two columns: zip_code and state. You cluster based on zip_code, which means that the data is sorted by zip_code when it is stored on disk. In the United States, there is a correlation between state and zip_code because zip code ranges are assigned geographically (00000 in the Northeast and 99999 in the Northwest). If you run a query that filters by state, even though you’re not filtering by the clustering column, BigQuery skips any data block that doesn’t have that state, and you’ll end up paying less for the query.

When a table is clustered, it allows BigQuery to apply a number of performance optimizations that are not possible with nonclustered tables. One of these optimizations is designed for star schemas, which let you filter based on constraints in a dimension table. For example, suppose that you have a fact table containing orders, which is clustered by customer_id, and a dimension table containing customers, and you run the following query:

SELECT o.*
FROM orders o 
JOIN customers c USING (customer_id)
WHERE c.name = "Changying Bao"

Ordinarily, this query would need to scan the full orders table in order to execute the join. But because the orders table is clustered by the column that is being used for the join, the righthand side of the join—querying customers and finding the matching customer IDs—is done first. Then the second part of the query just needs to look up the clustering column that matches the customer_id. From a bytes-scanned perspective, it is as if BigQuery ran the following queries in parallel:

// First look up the customer id.
// This scans only the small dimension table
SET id = SELECT customer_id FROM customers
WHERE c.name = "Changying Bao" 
// Next look up the customer from the orders table.
// This will filter by the cluster column,
// and so only needs to read a small amount of data.
SELECT * FROM orders WHERE customer_id=$id ;

In short, clustering is highly recommended if you want to reduce query costs and improve performance.