Tired of the limiting BigQuery console? Open your Jupyter notebook and start working with BigQuery using Python!
BiggerQuery lets you:
- Work with BigQuery using Python code.
- Create a workflow that you can automatically convert to an Airflow DAG.
- Implement a configurable environment for your workflows.
- Organize your data processing.
- Create a workflow from a Jupyter notebook.
- Work with BigQuery from any other environment.
- Run and schedule the Apache-Beam pipelines.
- Mix BigQuery, Python and Apache-Beam in your workflows.
BiggerQuery scales to your needs. It's very easy to start making queries and creating workflows. If needed, BiggerQuery lets you implement complex stuff (the Allegro experimentation platform was created using the BiggerQuery framework).
pip install biggerquery
pip install biggerquery[beam]
(if you want to use the Apache Beam)
BiggerQuery is compatible with Python >= 3.5.
We recommend using the Jupyter Lab to go through the examples. You can also run the examples as scripts, or from
your own Jupyter notebook. In those cases, you can authorize using pydata_google_auth
(look at the example below) or Google sdk.
Inside this repository you can find the file named 'MilitaryExpenditure.csv'. Use the script below to load the csv to the BigQuery table. You will use the created table to explore the BiggerQuery methods.
First of all, install the dependencies:
pip install biggerquery
pip install pydata_google_auth
Then, fill up the PROJECT_ID and DATA_PATH:
PROJECT_ID = 'put-you-project-id-here'
DATA_PATH = '/path/to/json/file/311_requests.csv'
import biggerquery as bgq
import pydata_google_auth
import pandas as pd
credentials = pydata_google_auth.get_user_credentials(['https://www.googleapis.com/auth/bigquery'])
dataset = bgq.Dataset(
project_id=PROJECT_ID,
dataset_name='external_data',
credentials=credentials)
df = pd.read_csv(DATA_PATH, dtype={
'street_number': str,
'state_plane_x_coordinate': str
})
load_table = dataset.load_table_from_dataframe('311_requests', df, partitioned=False)
load_table.run()
import biggerquery as bgq
import pydata_google_auth
credentials = pydata_google_auth.get_user_credentials(['https://www.googleapis.com/auth/bigquery'])
dataset = bgq.Dataset(
project_id='put-you-project-id-here',
dataset_name='biggerquery_cheatsheet',
credentials=credentials)
import biggerquery as bgq
dataset = bgq.Dataset(
project_id='put-you-project-id-here',
dataset_name='biggerquery_cheatsheet',
internal_tables=['request_aggregate'])
create_table = dataset.create_table("""
CREATE TABLE IF NOT EXISTS request_aggregate (
batch_date TIMESTAMP,
request_count INT64)
PARTITION BY DATE(batch_date)""")
create_table.run()
import biggerquery as bgq
PROJECT_ID = 'put-you-project-id-here'
dataset = bgq.Dataset(
project_id=PROJECT_ID,
dataset_name='biggerquery_cheatsheet',
external_tables={
'311_requests': '{}.external_data.311_requests'.format(PROJECT_ID)
})
select_requests = dataset.collect("""
SELECT *
FROM `{311_requests}`
WHERE DATE(TIMESTAMP(created_date)) = "{dt}"
LIMIT 1000
""")
requests_df = select_requests.run('2014-05-21')
print(requests_df)
import biggerquery as bgq
PROJECT_ID = 'put-you-project-id-here'
dataset = bgq.Dataset(
project_id=PROJECT_ID,
dataset_name='biggerquery_cheatsheet',
external_tables={
'311_requests': '{}.external_data.311_requests'.format(PROJECT_ID)
})
dry_select = dataset.dry_run("""
SELECT *
FROM `{311_requests}`
WHERE DATE(TIMESTAMP(created_date)) = "{dt}"
LIMIT 1000
""")
print(dry_select.run('2014-05-21'))
import biggerquery as bgq
PROJECT_ID = 'put-you-project-id-here'
dataset = bgq.Dataset(
project_id=PROJECT_ID,
dataset_name='biggerquery_cheatsheet',
external_tables={
'311_requests': '{}.external_data.311_requests'.format(PROJECT_ID)
},
internal_tables=['request_aggregate'])
create_table = dataset.create_table("""
CREATE TABLE IF NOT EXISTS request_aggregate (
batch_date TIMESTAMP,
request_count INT64)
PARTITION BY DATE(batch_date)""").run()
write_truncate_daily_request_count = dataset.write_truncate('request_aggregate', """
WITH batched_requests as (
SELECT
DATE(TIMESTAMP(created_date)) as batch_date,
*
FROM `{311_requests}`
WHERE DATE(TIMESTAMP(created_date)) = "{dt}"
)
SELECT
TIMESTAMP(batch_date) as batch_date,
count(*) as request_count
FROM `batched_requests`
WHERE DATE(TIMESTAMP(created_date)) = "{dt}"
GROUP BY batch_date
""")
write_truncate_daily_request_count.run('2014-05-21')
import biggerquery as bgq
PROJECT_ID = 'put-you-project-id-here'
dataset = bgq.Dataset(
project_id=PROJECT_ID,
dataset_name='biggerquery_cheatsheet',
external_tables={
'311_requests': '{}.external_data.311_requests'.format(PROJECT_ID)
},
internal_tables=['request_aggregate_tmp'])
write_tmp_daily_request_count = dataset.write_tmp('request_aggregate_tmp', """
WITH batched_requests as (
SELECT
DATE(TIMESTAMP(created_date)) as batch_date,
*
FROM `{311_requests}`
WHERE DATE(TIMESTAMP(created_date)) = "{dt}"
)
SELECT
TIMESTAMP(batch_date) as batch_date,
count(*) as request_count
FROM `batched_requests`
WHERE DATE(TIMESTAMP(created_date)) = "{dt}"
GROUP BY batch_date
""")
write_tmp_daily_request_count.run('2014-05-21')
import biggerquery as bgq
import pandas as pd
PROJECT_ID = 'put-you-project-id-here'
dataset = bgq.Dataset(
project_id=PROJECT_ID,
dataset_name='biggerquery_cheatsheet',
external_tables={
'311_requests': '{}.external_data.311_requests'.format(PROJECT_ID)
},
internal_tables=['request_aggregate'])
create_table = dataset.create_table("""
CREATE TABLE IF NOT EXISTS request_aggregate (
batch_date TIMESTAMP,
request_count INT64)
PARTITION BY DATE(batch_date)""").run()
load_df = dataset.load_table_from_dataframe('request_aggregate', pd.DataFrame([{
'batch_date': pd.Timestamp('2017-01-01T12'),
'request_count': 200
}]))
load_df.run('2017-01-01')
Create an empty notebook and add the following processing logic:
import biggerquery as bgq
PROJECT_ID = 'put-you-project-id-here'
dataset = bgq.Dataset(
project_id=PROJECT_ID,
dataset_name='biggerquery_cheatsheet',
external_tables={
'311_requests': '{}.external_data.311_requests'.format(PROJECT_ID)
},
internal_tables=['request_aggregate'])
create_table = dataset.create_table("""
CREATE TABLE IF NOT EXISTS request_aggregate (
batch_date TIMESTAMP,
request_count INT64)
PARTITION BY DATE(batch_date)""")
write_truncate_daily_request_count = dataset.write_truncate('request_aggregate', """
WITH batched_requests as (
SELECT
DATE(TIMESTAMP(created_date)) as batch_date,
*
FROM `{311_requests}`
WHERE DATE(TIMESTAMP(created_date)) = "{dt}"
)
SELECT
TIMESTAMP(batch_date) as batch_date,
count(*) as request_count
FROM `batched_requests`
WHERE DATE(TIMESTAMP(created_date)) = "{dt}"
GROUP BY batch_date
""")
workflow_v1 = bgq.Workflow(definition=[
create_table.to_job(),
write_truncate_daily_request_count.to_job()
])
Next, create another notebook and add the following code that will generate the Airflow DAG:
import biggerquery as bgq
bgq.build_dag_from_notebook('/path/to/your/notebook.ipynb', 'workflow_v1', start_date='2014-05-21')
After you run the code above, you will get a zipped Airflow DAG that you can deploy. The easiest way to deploy a DAG is by using the Cloud Composer.
If you want to wait for some data to appear before you start a processing, you can use sensor component:
import biggerquery as bgq
PROJECT_ID = 'put-you-project-id-here'
dataset = bgq.Dataset(
project_id=PROJECT_ID,
dataset_name='biggerquery_cheatsheet',
external_tables={
'311_requests': '{}.external_data.311_requests'.format(PROJECT_ID)
},
internal_tables=['request_aggregate'])
wait_for_requests = bgq.sensor_component(
'311_requests',
where_clause="DATE(TIMESTAMP(created_date)) = DATE(TIMESTAMP_ADD(TIMESTAMP('{dt}'), INTERVAL -24 HOUR))",
ds=dataset)
workflow_v2 = bgq.Workflow(definition=[wait_for_requests.to_job()])
# Should raise ValueError because there is no data for '2090-01-01'
workflow_v2.run('2090-01-01')
If you want to write you own component, you can do it by writing a function:
import biggerquery as bgq
PROJECT_ID = 'put-you-project-id-here'
dataset = bgq.Dataset(
project_id=PROJECT_ID,
dataset_name='biggerquery_cheatsheet',
external_tables={
'311_requests': '{}.external_data.311_requests'.format(PROJECT_ID)
},
internal_tables=['request_aggregate'])
def is_table_ready(df):
return df.iloc[0]['table_ready']
@bgq.component(ds=dataset)
def wait_for_requests(ds):
result = ds.collect('''
SELECT count(*) > 0 as table_ready
FROM `{311_requests}`
WHERE DATE(TIMESTAMP(created_date)) = DATE(TIMESTAMP_ADD(TIMESTAMP('{dt}'), INTERVAL -24 HOUR))
''')
if not is_table_ready(result):
raise ValueError('311_requests is not ready')
workflow_v2 = bgq.Workflow(definition=[wait_for_requests.to_job()])
# Should raise ValueError because there is no data for '2090-01-01'
workflow_v2.run('2090-01-01')
BiggerQuery offers a cli (command-line interface) that lets you run or deploy jobs and workflows directly from your terminal. The main commands are:
run
- lets you run a job or a workflow,
To run any command, start with bgq
and command name. To ask for help, use bgq -h
or bgq <command name> -h
.
run
command lets you run a job or a workflow. Here are a few examples how it can be used:
bgq run --workflow workflowId
bgq run --workflow workflowId --runtime '2020-01-01 00:00:00' --config prod
bgq run --job jobId
bgq run --job jobId --runtime '2020-01-01 00:00:00'
bgq run --job jobId --runtime '2020-01-01 00:00:00' --config dev
Run command requires you to provide one of those two parameters:
--job <job id>
- use it to run a job by its id. You can set job id by settingid
field in the object representing this job.--workflow <workflow id>
- use it to run a workflow by its id. You can set workflow id using named parameterworkflow_id
(bgq.Workflow(workflow_id="YOUR_ID", ...)
). In both cases, id needs to be set and unique.
Run command also allows the following optional parameters:
--runtime <runtime in format YYYY-MM-DD hh:mm:ss>
- use it to set the date and time when this job or workflow should be started. Example value:2020-01-01 00:12:00
. The default is now.--config <runtime>
- use it to configure environment name that should be used. Example:dev
,prod
. If not set, the default Config name will be used. This env name is applied to all biggerquery.Config objects that are defined by individual workflows as well as to deployment_config.py.--project_package <project_package>
- use it to set the main package of your project, only when project_setup.PROJECT_NAME not found. Example:logistics_tasks
. The value does not affect when project_setup.PROJECT_NAME is set. Otherwise, it is required.
Inside this repository, you can find the BiggerQuery tutorial. We recommend using the GCP Jupyter Lab to go through the tutorial. It takes a few clicks to set up.