Dataset information

Given the dsinfo object, it is possible to extract different attributes of the dataset. For example,

print(dsinfo.dataset_id)
print(dsinfo.created)

on the ch04 object yields

ch04
2019-01-26 00:41:01.350000+00:00

whereas

print('{} created on {} in {}'.format(
      dsinfo.dataset_id, dsinfo.created, dsinfo.location))

for the bigquery-public-data.london_bicycles dataset yields the following:

london_bicycles created on 2017-05-25 13:26:18.055000+00:00 in EU

It is also possible to examine the access controls on the dataset using the dsinfo object. For example, we could find which roles are granted READER access to the london_bicycles dataset using this:

for access in dsinfo.access_entries:
    if access.role == 'READER':
        print(access)

This yields the following:

<AccessEntry: role=READER, specialGroup=allAuthenticatedUsers>
<AccessEntry: role=READER, domain=google.com>
<AccessEntry: role=READER, specialGroup=projectReaders>

It is because all authenticated users are granted access to the dataset (see the first line in the preceding example) that we have been able to query the dataset in previous chapters.

Creating a dataset

To create a dataset named ch05 if it doesn’t already exist, use this:

dataset_id = "{}.ch05".format(PROJECT)
ds = bq.create_dataset(dataset_id, exists_ok=True)

By default, the dataset is created in the US. To create the dataset in another location—for example, the EU—create a local Dataset object (we’re calling it dsinfo), set its location attribute, and then invoke create_dataset on the client object using this Dataset (instead of the dataset_id, as in the previous code snippet):

dataset_id = "{}.ch05eu".format(PROJECT)
dsinfo = bigquery.Dataset(dataset_id)
dsinfo.location = 'EU'
ds = bq.create_dataset(dsinfo, exists_ok=True)

Deleting a dataset

To delete a dataset named ch05 in the project passed into the Client, do the following:

bq.delete_dataset('ch05', not_found_ok=True)

To delete a dataset in a different project, qualify the dataset name by the project name:

bq.delete_dataset('{}.ch05'.format(PROJECT), not_found_ok=True)

Modifying attributes of a dataset

To modify information about a dataset, modify the dsinfo object locally by setting the description attribute and then invoke update_dataset on the client object to update the BigQuery service:

dsinfo = bq.get_dataset("ch05")
print(dsinfo.description)
dsinfo.description = "Chapter 5 of BigQuery: The Definitive Guide"
dsinfo = bq.update_dataset(dsinfo, ['description'])
print(dsinfo.description)

The first print in the preceding snippet prints out None because the dataset ch05 was created without any description. After the update_dataset call, the dataset in BigQuery sports a new description:

None
Chapter 5 of BigQuery: The Definitive Guide

Changing tags, access controls, and so on of a dataset works similarly. For example, to give one of our colleagues access to the ch05 dataset, we could do the following:

dsinfo = bq.get_dataset("ch05")
entry = bigquery.AccessEntry(
    role="READER",
    entity_type="userByEmail",
    entity_id="[email protected]",
)
if entry not in dsinfo.access_entries:
  entries = list(dsinfo.access_entries)
  entries.append(entry)
  dsinfo.access_entries = entries
  dsinfo = bq.update_dataset(dsinfo, ["access_entries"]) # API request
else:
  print('{} already has access'.format(entry.entity_id))
print(dsinfo.access_entries)

In this code, we create an entry for a user, and if the user doesn’t already have some sort of access to the dataset, we get the current set of access entries, append the new entry, and update the dataset with the new list.

Obtaining table properties

In the previous code snippet, we got the table_id from the table object. Besides the table_id, other attributes of the table are available: the number of rows, the descriptions, tags, the schema, and more.

table = bq.get_table(
   "bigquery-public-data.london_bicycles.cycle_stations")
print('{} rows in {}'.format(table.num_rows,
table.table_id))

787 rows in cycle_stations

For example, we can search the schema for columns whose name contains a specific substring (count) using the table object:

table = bq.get_table(
          "bigquery-public-data.london_bicycles.cycle_stations")
for field in table.schema:
  if 'count' in field.name:
    print(field)

Here’s the result:

SchemaField('bikes_count', 'INTEGER', 'NULLABLE', '', ())
SchemaField('docks_count', 'INTEGER', 'NULLABLE', '', ())

Of course, rather than hand-roll this sort of search, it would be better to use INFORMATION_SCHEMA (covered in Chapter 8) or Data Catalog.

Deleting a table

Deleting a table is similar to deleting a dataset, and if desired, you can ignore the error thrown if the table doesn’t exist:

bq.delete_table('ch05.temp_table', not_found_ok=True)

Creating an empty table

Creating an empty table is similar to creating a dataset, and if desired, you can ignore the exception thrown if the dataset already exists:

table_id = '{}.ch05.temp_table'.format(PROJECT)
table = bq.create_table(table_id, exists_ok=True)

Updating a table’s schema

Of course, you don’t usually want to create empty tables. You want to create an empty table with a schema and insert some rows into it. Because the schema is part of the attributes of the table, you can update the schema of the empty table similarly to the way you updated the access controls of the dataset. You get the table, modify the table object locally, and then update the table using the modified object to specify what aspects of the table object are being updated:

schema = [
  bigquery.SchemaField("chapter", "INTEGER", mode="REQUIRED"),
  bigquery.SchemaField("title", "STRING", mode="REQUIRED"),
]
table_id = '{}.ch05.temp_table'.format(PROJECT)
table = bq.get_table(table_id)
print(table.etag)
table.schema = schema
table = bq.update_table(table, ["schema"])
print(table.schema)
print(table.etag)

To prevent race conditions, BigQuery tags the table with each update. So when you get the table information using get_table, the table object includes an etag. When you upload a modified schema using update_table, this update succeeds only if your etag matches that of the server. The returned table object has the new etag. You can turn off this behavior and force an update by setting table.etag to None.

When a table is empty, you can change the schema to anything you want. But when there is data in the table, any schema changes must be compatible with the existing data in the table. You can add new fields (as long as they are NULLABLE), and you can relax constraints from REQUIRED to NULLABLE.

After this code is run, we can check in the BigQuery web UI that the newly created table has the correct schema, as depicted in Figure 5-2.

Figure 5-2. Schema of the newly created table

Inserting rows into a table

After you have a table with a schema, you can insert rows into the table using the client. The rows consist of Python tuples in the same order as defined in the schema:

rows = [
  (1, u'What is BigQuery?'),
  (2, u'Query essentials'),
]
errors = bq.insert_rows(table, rows)

The errors list will be empty if all rows were successfully inserted. If, however, you had passed in a noninteger value for the chapter field

rows = [
  ('3', u'Operating on data types'),
  ('wont work', u'This will fail'),
  ('4', u'Loading data into BigQuery'),
]
errors = bq.insert_rows(table, rows)
print(errors)

you will get an error whose reason is invalid on index=1 (the second row; this is 0-based), location=chapter:

{'index': 1, 'errors': [{'reason': 'invalid', 'debugInfo': '', 'message': 'Cannot
convert value to integer (bad value):wont work', 'location': 'chapter'}]}

Because BigQuery treats each user request as atomic, none of the three rows will be inserted. On the other rows, you will get an error whose reason is stopped:

{'index': 0, 'errors': [{'reason': 'stopped', 'debugInfo': '', 'message': '',
'location': ''}]}

In the BigQuery web UI, shown in Figure 5-3, the table details show that the inserted rows are in the streaming buffer, but the two inserted rows are not reflected in the table’s number of rows.

Figure 5-3. Newly inserted rows are in the streaming buffer and are not yet reflected in the number of rows shown in the “Table info” section

rows = [
  (1, u'What is BigQuery?'),
  (2, u'Query essentials'),
]
print(table.table_id, table.num_rows)
errors = bq.insert_rows(table, rows)
print(errors)
table = bq.get_table(table_id)
print(table.table_id, table.num_rows) # DELAYED

Queries on the table will reflect the two rows in the streaming buffer:

SELECT DISTINCT(chapter) FROM ch05.temp_table

This shows that there are two chapters in the table:

Creating an empty table with schema

Instead of creating a table and then updating this schema, a better idea is to provide the schema at the time of table creation:

schema = [
  bigquery.SchemaField("chapter", "INTEGER", mode="REQUIRED"),
  bigquery.SchemaField("title", "STRING", mode="REQUIRED"),
]
table_id = '{}.ch05.temp_table2'.format(PROJECT)
table = bigquery.Table(table_id, schema)
table = bq.create_table(table, exists_ok=True)
print('{} created on {}'.format(table.table_id, table.created))
print(table.schema)

The created table contains the desired schema:

temp_table2 created on 2019-03-03 19:30:18.324000+00:00
[SchemaField('chapter', 'INTEGER', 'REQUIRED', None, ()), SchemaField('title',
'STRING', 'REQUIRED', None, ())]

The table created is empty, and so we’d use this technique if we are going to do a streaming insert of rows into the table. What if, though, you already have the data in a file and you want simply to create a table and initialize it with data from that file? In that case, load jobs are much more convenient. Unlike streaming inserts, loads do not incur BigQuery charges.

The BigQuery Python client supports three methods of loading data: from a pandas DataFrame, from a URI, or from a local file. Let’s look at these next.

Loading a pandas DataFrame

pandas is an open source library that provides data structures and data analysis tools for the Python programming language. The BigQuery Python library supports directly loading data from an in-memory pandas DataFrame.  Because pandas DataFrames can be constructed from in-memory data structures and provide a wide variety of transformations, using pandas provides the most convenient way to load data from Python applications. For example, to create a DataFrame from an array of tuples, you can do the following:

import pandas as pd
 data = [
  (1, u'What is BigQuery?'),
  (2, u'Query essentials'),
]
df = pd.DataFrame(data, columns=['chapter', 'title'])

After you have created the DataFrame, you can load the data within it into a BigQuery table using the following:⁶

table_id = '{}.ch05.temp_table3'.format(PROJECT)
job = bq.load_table_from_dataframe(df, table_id)
job.result() # blocks and waits
print("Loaded {} rows into {}".format(job.output_rows, 
                                             tblref.table_id))

Because load jobs can potentially be long running, the load_table_ function returns a job object that you can use either to poll, using the job.done() method, or to block and wait, using job.result().

If the table already exists, the load job will append to the existing table as long as the data you are loading matches the existing schema. If the table doesn’t exist, a new table is created with schema that is inferred from the pandas DataFrame.⁷ You can change this behavior by specifying a load configuration:

from google.cloud.bigquery.job 
      import LoadJobConfig, WriteDisposition, CreateDisposition
load_config = LoadJobConfig(
  create_disposition=CreateDisposition.CREATE_IF_NEEDED,
  write_disposition=WriteDisposition.WRITE_TRUNCATE)
job = bq.load_table_from_dataframe(df, table_id, 
                                   job_config=load_config)

The combination of CreateDisposition and WriteDisposition controls the behavior of the load operation, as is shown in Table 5-2.

Loading from a URI

It is possible to load a BigQuery table directly from a file whose Google Cloud URI is known. In addition to Cloud Datastore backups and HTTP URLs referring to Cloud Bigtable, Google Cloud Storage wildcard patterns are also supported.⁸ We can, therefore, load the college scorecard comma-separated values (CSV) file that we used in the previous chapter by using the following:

job_config = bigquery.LoadJobConfig()
job_config.autodetect = True
job_config.source_format = bigquery.SourceFormat.CSV
job_config.null_marker = 'NULL'
uri = "gs://bigquery-oreilly-book/college_scorecard.csv"
table_id = '{}.ch05.college_scorecard_gcs'.format(PROJECT)
job = bq.load_table_from_uri(uri, table_id, job_config=job_config)

You can set all of the options that we considered in Chapter 4 (on loading data) by using the JobConfig flags. Here, we are using autodetect, specifying that the file format is CSV and that the file uses a nonstandard null marker before loading the file from the URI specified.

Even though you can block for the job to finish as you did in the previous section, you can also poll the job every 0.1 seconds and get the table details only after the load job is done:

while not job.done():
  print('.', end='', flush=True)
  time.sleep(0.1)
print('Done')
table = bq.get_table(tblref)
print("Loaded {} rows into {}.".format(table.num_rows, table.table_id))

After a few dots to represent the wait state, you get back the number of rows loaded (7,175).

Loading from a local file

To load from a local file, create a file object and use load_table_from_file:

with gzip.open('../04_load/college_scorecard.csv.gz') as fp:
  job = bq.load_table_from_file(fp, tblref, job_config=job_config)

In other respects, this is similar to loading from a URI.

Copying a table

You can copy a table from one dataset to another by using the copy_table method:

source_tbl = 'bigquery-public-data.london_bicycles.cycle_stations'
dest_tbl = '{}.ch05eu.cycle_stations_copy'.format(PROJECT)
job = bq.copy_table(source_tbl, dest_tbl, location='EU')
job.result() # blocks and waits
dest_table = bq.get_table(dest_tbl)
print(dest_table.num_rows)

Note that we are copying the London cycle_stations data to a dataset (ch05eu) that we created in the EU. Note also that we are making sure to copy tables only within the same region.

Extracting data from a table

We can export data from a table to a file in Google Cloud Storage using the extract_table method:

source_tbl = 'bigquery-public-data.london_bicycles.cycle_stations'
dest_uri = 'gs://{}/tmp/exported/cycle_stations'.format(BUCKET)
config = bigquery.job.ExtractJobConfig(
  destination_format =
        bigquery.job.DestinationFormat.NEWLINE_DELIMITED_JSON)
job = bq.extract_table(source_tbl, dest_uri, 
                        location='EU', job_config=config)
job.result() # blocks and waits

If the table is sufficiently large, the output will be sharded into multiple files. As of this writing, extraction formats that are supported include CSV, Avro, and newline-delimited JSON. As with copying tables, make sure to export to a bucket in the same location as the dataset.⁹ Of course, after you export a table, Google Cloud Storage charges will begin to accrue for the output files.

Browsing the rows of a table

In the BigQuery web UI, you have the ability to preview a table without incurring querying charges. The same capability is available via the REST API as tabledata.list, and consequently through the Python API.

To list an arbitrary five rows from the cycle_stations table,¹⁰ you could do the following:

table_id = 'bigquery-public-data.london_bicycles.cycle_stations'
table = bq.get_table(table_id)
rows = bq.list_rows(table, 
                          start_index=0, 
                          max_results=5)

Omitting the start_index and max_results allows you to get all of the rows in the table:

rows = bq.list_rows(table)

Of course, the table needs to be small enough to fit into memory. If that is not the case, you can paginate through the entire table, processing the table in chunks:

page_size = 10000
row_iter = bq.list_rows(table,
                        page_size=page_size)
for page in row_iter.pages:
  rows = list(page)
  # do something with rows ...
  print(len(rows))

Instead of getting all of the fields, you can select the id field and any columns whose name includes the substring count by doing the following:

fields = [field for field in table.schema 
             if 'count' in field.name or field.name == 'id']
rows = bq.list_rows(table, 
                    start_index=300, 
                    max_results=5, 
                    selected_fields=fields)

You can then format the resulting rows to have a fixed width of 10 characters using the following:

fmt = '{!s:<10} ' * len(rows.schema)
print(fmt.format(*[field.name for field in rows.schema]))
for row in rows:
  print(fmt.format(*row))

This produces the following result:

id         bikes_count docks_count 
658        20         30         
797        20         30         
238        21         32         
578        22         32         
477        26         36

Dry run

Before actually executing the query, it is possible to do a dry run to obtain an estimate of how much data will be processed by the query:¹¹

config = bigquery.QueryJobConfig()
config.dry_run = True
job = bq.query(query, location='EU', job_config=config)
print("This query will process {} bytes."
                .format(job.total_bytes_processed))

When we ran the preceding code, it returned the following:

This query will process 903989528 bytes.
Your result might be somewhat different given that this table is refreshed with 
new data as it is made available.

Sometimes it is impossible to compute the bytes processed ahead of time without actually running the query. In such cases, the dry run returns either zero or an upper-bound estimate. This happens in two situations: when querying a federated table (for which the data is stored outside BigQuery; see Chapter 4) and when querying a clustered table (see Chapter 7). In the case of federated tables, the dry run will report 0 bytes, and in the case of clustered tables, BigQuery will attempt to calculate the worst-case scenario and report that number. In either case, though, when actually performing the query, you’ll be billed only for the data that actually needs to be read.

Executing the query

To execute the query, simply start to iterate over the job object. The job will be launched, and pages of results will be retrieved as you iterate over the job using the for loop:

job = bq.query(query, location='EU')
fmt = '{!s:<40} {:>10d} {:>10d}'
for row in job:
    fields = (row['start_station_name'], 
             (int)(0.5 + row['duration']), 
             row['num_trips']) 
    print(fmt.format(*fields))

Given a row, it is possible to obtain the value for any of the columns in the result set using the aliased name of the column in the SELECT (look at how the column num_trips appears in the result set).

The formatted result of the query is as follows:

Belgrove Street, King's Cross                  1011     234458
Hyde Park Corner, Hyde Park                    2783     215629
Waterloo Station 3, Waterloo                    866     201630
Black Lion Gate, Kensington Gardens            3588     161952
Albert Gate, Hyde Park                         2359     155647
Waterloo Station 1, Waterloo                    992     145910
Wormwood Street, Liverpool Street               976     119447
Hop Exchange, The Borough                      1218     115135
Wellington Arch, Hyde Park                     2276     110260
Triangle Car Park, Hyde Park                   2233     108347

Creating a pandas DataFrame

Earlier in this section, we saw how to load a BigQuery table from a pandas DataFrame. It is also possible to execute a query and get the results back as a pandas DataFrame, thus using BigQuery as a highly distributed and scalable intermediate step in a data science workflow:

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
"""
df = bq.query(query, location='EU').to_dataframe()
print(df.describe())

This code uses the pandas describe() functionality to print out the distribution of the numeric columns in the result set:

          duration      num_trips
count   880.000000     880.000000
mean   1348.351153   27692.273864
std     434.057829   23733.621289
min       0.000000       1.000000
25%    1078.684974   13033.500000
50%    1255.889223   23658.500000
75%    1520.504055   35450.500000
max    4836.380090  234458.000000

Thus there are 880 stations in total, with an average of 27,692 trips starting at each station, although there is a station with only one trip and a station with 234,458 trips. The median station has supported 23,658 rides, and the majority of stations have had between 13,033 and 35,450 rides.

Parameterized queries

The queries do not need to be static strings. Instead, you can parameterize them, so that the query parameters are specified at the time the query job is created. Here is an example of a query that finds the total number of trips that were longer than a specific duration. The actual threshold, min_duration, will be specified at the time the query is run:

query2 = """
SELECT 
  start_station_name 
  , COUNT(duration) as num_trips
FROM `bigquery-public-data`.london_bicycles.cycle_hire 
WHERE duration >= @min_duration
GROUP BY start_station_name 
ORDER BY num_trips DESC
LIMIT 10
"""

The @ symbol identifies min_duration as a parameter to the query. A query can have any number of such named parameters.

query2 = """
SELECT 
  start_station_name 
  , COUNT(duration) as num_trips
FROM `bigquery-public-data`.london_bicycles.cycle_hire 
WHERE duration >= {}
GROUP BY start_station_name 
ORDER BY num_trips DESC
LIMIT 10
""".format(min_duration)

When executing a query that has named parameters, you need to supply a job_config with those parameters:

config = bigquery.QueryJobConfig()
config.query_parameters = [
  bigquery.ScalarQueryParameter('min_duration', "INT64", 600)
]
job = bq.query(query2, location='EU', job_config=config)

Here, we are specifying that we want to retrieve the number of trips over 600 seconds in duration.

As before, iterating over the job will allow you to retrieve the rows, and each row functions like a dictionary of column names to values:

fmt = '{!s:<40} {:>10d}'
for row in job:
    fields = (row['start_station_name'],
              row['num_trips']) 
    print(fmt.format(*fields))

Running this code yields the following:

Hyde Park Corner, Hyde Park                  203592
Belgrove Street, King's Cross                168110
Waterloo Station 3, Waterloo                 148809
Albert Gate, Hyde Park                       145794
Black Lion Gate, Kensington Gardens          137930
Waterloo Station 1, Waterloo                 106092
Wellington Arch, Hyde Park                   102770
Triangle Car Park, Hyde Park                  99368
Wormwood Street, Liverpool Street             82483
Palace Gate, Kensington Gardens               80342

In this section, we covered how to programmatically invoke BigQuery operations, whether they involve table or dataset manipulation, querying data, or streaming inserts. The programmatic APIs, especially the Google Cloud Client Library, are what you would use whenever you are building applications that need to access BigQuery.

However, in some specific instances, there are higher-level abstractions available. We cover these in the next section.