> **Can't find what you're looking for?** Use `search_docs` on the docs MCP server at `https://docs.ampdb.io/api/mcp` to find what you need. # BigQuery Loader The BigQuery loader writes query results into tables in [BigQuery](https://cloud.google.com/bigquery), Google Cloud's data warehouse. Choose it to analyze Amp's data with BigQuery's SQL, or with the tools you already connect to it, like Looker Studio. It creates the dataset and the tables it needs, and loads rows one of two ways: with BigQuery's streaming inserts (the default), or with load jobs that go through a Cloud Storage bucket. Load jobs are the ones that work with [reorgs](/sdks/python/loaders/bigquery#streaming-and-reorgs). [Loaders](/sdks/python/loaders) explains what every loader has in common: named connections, load options, and the `LoadResult` each load returns. ## Prerequisites * **Amp running, with the Quickstart's dataset.** The examples query `"acme/eth_mainnet"`, which the [Quickstart](/get-started/quickstart) deploys, through solo mode's query server at `grpc://localhost:1602`. * **The Python SDK**, installed as [Installation](/sdks/python#installation) shows. * **Google's BigQuery client**, which the SDK doesn't install, and the Cloud Storage client for the `load_job` method: :::code-group ```bash [pip] pip install google-cloud-bigquery # Only for loading_method='load_job' pip install google-cloud-storage ``` ```bash [uv] uv add google-cloud-bigquery # Only for loading_method='load_job' uv add google-cloud-storage ``` ::: * **A Google Cloud project** with the BigQuery API turned on. Set its ID in an environment variable: ```bash export BIGQUERY_PROJECT=… ``` * **Credentials** that can create datasets and tables, and run jobs, in that project (the BigQuery Data Editor and BigQuery Job User roles), and for `load_job`, write and delete objects in the bucket (Storage Object Admin). Either sign in with the [gcloud CLI](https://cloud.google.com/sdk/docs/install), whose credentials the loader finds on its own: ```bash gcloud auth application-default login ``` Or download a service account's key, and point to it: ```bash export GOOGLE_APPLICATION_CREDENTIALS=… ``` * **A Cloud Storage bucket**, only for `load_job`: the loader puts a file there for each batch, and deletes it once BigQuery has loaded it. Use one in the same location as the dataset. ```bash export BIGQUERY_GCS_BUCKET=… ``` ## Load a table This script loads the Quickstart's 100 blocks into a table named `blocks`, in a BigQuery dataset named `amp`: ```python [load_bigquery.py] import os from amp import Client client = Client('grpc://localhost:1602') client.configure_connection( 'my_bigquery', 'bigquery', { 'project': os.environ['BIGQUERY_PROJECT'], 'dataset': 'amp', 'location': 'US', }, ) result = client.sql('SELECT block_num, timestamp, hash, gas_used FROM "acme/eth_mainnet".blocks').load( connection='my_bigquery', destination='blocks', read_all=True ) print(result) ``` ```bash python load_bigquery.py ``` When it succeeds, it prints `✅ Loaded 100 rows to blocks`, with the time the load took. When it fails, it prints `❌ Failed to load to blocks:` and the error: a load doesn't raise exceptions, it returns them in its `LoadResult`. * `configure_connection()` names a connection, `my_bigquery`, and says which loader it uses, `bigquery`. Every [option](/sdks/python/loaders/bigquery#configuration) goes in its dictionary. * The loader creates the `amp` dataset, in the `US` location, if it doesn't exist, then the `blocks` table. * `read_all=True` loads the whole result at once. Without it, the loader loads it batch by batch. :::warning `configure_connection()` saves the connection's configuration in plain text in `~/.flight_sql_connections.json`, so other scripts can use it by name. Pass a key file's path in `credentials_json`, or use the gcloud CLI's credentials, rather than a key's contents in `credentials_dict`. Set the `FLIGHT_SQL_CONNECTIONS` environment variable to keep the file elsewhere. ::: Then query it in BigQuery: ```sql SELECT block_num, timestamp, gas_used FROM amp.blocks ORDER BY block_num DESC LIMIT 5; ``` ## Configuration `project` and `dataset` are required. Any key not listed here fails the load (see [Troubleshooting](/sdks/python/loaders/bigquery#troubleshooting)). The `state` and `resilience` keys, which every loader takes, are on [Loaders](/sdks/python/loaders). | Option | Type | Default | Description | | ------------------- | ---------- | ---------------------------------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | `project` | string | required | The Google Cloud project's ID, like `'my-project'`. | | `dataset` | string | required | The BigQuery dataset to load into, like `'amp'`. The loader creates it if it doesn't exist. | | `location` | string | none | Where the loader creates the dataset, and runs jobs, like `'US'`, `'EU'`, or `'us-central1'`. Without it, BigQuery uses its default, `US`. | | `loading_method` | string | `'insert_rows'` | How rows get into BigQuery: `'insert_rows'` or `'load_job'` (see [Loading methods](/sdks/python/loaders/bigquery#loading-methods)). | | `gcs_bucket` | string | none | The Cloud Storage bucket `load_job` stages files in, like `'my-staging-bucket'`. Required with `load_job`. | | `gcs_prefix` | string | `'amp-staging'` | The folder in the bucket the files go in, like `'staging/amp'`. Each file goes in a subfolder named after the table's full ID. | | `batch_size` | integer | `10000` | With `insert_rows`: the most rows to send in one request, up to BigQuery's limit of 10,000, like `5000`. Lower it for wide rows, since a request is also limited to 10 MB. | | `credentials_json` | string | none | The path to a service account's key file, like `'/secrets/loader.json'`. | | `credentials_dict` | dictionary | none | A service account's key, as a dictionary, like `json.load(open('loader.json'))`. It takes precedence over `credentials_json`. | | `connection_params` | dictionary | `{}` | Other arguments for BigQuery's `Client`, like `{'client_options': {'api_endpoint': 'https://bigquery.googleapis.com'}}`. | | `state` | dictionary | `{'enabled': True, 'storage': 'memory'}` | Where the loader records the block ranges it loaded. [Loaders](/sdks/python/loaders) lists its keys. | Without `credentials_dict` or `credentials_json`, the loader uses Google's [Application Default Credentials](https://cloud.google.com/docs/authentication/application-default-credentials): `GOOGLE_APPLICATION_CREDENTIALS`, then the gcloud CLI's sign-in, then the service account of the Google Cloud machine it runs on. ## What the loader creates When it connects, the loader creates the dataset, if it doesn't exist. Before the first batch, it creates the destination table, `project.dataset.table`, if it doesn't exist. Each column gets the closest BigQuery type: | Amp (Arrow) type | BigQuery type | | -------------------------------------- | --------------------------------------------- | | `Int8` to `Int64`, `UInt8` to `UInt32` | `INT64` | | `UInt64` | `NUMERIC`, which holds every `UInt64` value | | `Float16` to `Float64` | `FLOAT64` | | `Decimal128(p, s)`, `Decimal256(p, s)` | `NUMERIC` up to 29 digits, `BIGNUMERIC` above | | `Utf8` | `STRING` | | `Binary`, `FixedSizeBinary(n)` | `BYTES` | | `Boolean` | `BOOL` | | `Timestamp` with a time zone | `TIMESTAMP`, in UTC | | `Timestamp` without a time zone | `DATETIME` | | `Date32`, `Time64` | `DATE`, `TIME` | | `Duration` | `INT64`, in microseconds | | Lists, structs, maps | `STRING`, holding the value as JSON | Columns that can't be null are `REQUIRED`. The loader also adds an `_amp_batch_id` column (`STRING`), which identifies the block range each row came from, so a reorg can remove the rows of the blocks it replaces. It's `NULL` for rows loaded without streaming. The table is clustered on it, which makes those deletions cheaper. A table that already exists must have an `_amp_batch_id` column, or the load fails (see [Troubleshooting](/sdks/python/loaders/bigquery#troubleshooting)). ## Loading methods | Method | How it loads each batch | Choose it for | | ------------- | ---------------------------------------------------------------------------------------------------------------------- | ------------------------------------------------------------------------ | | `insert_rows` | Sends its rows to BigQuery's streaming insert API, in requests of up to `batch_size` rows. | Batch loads, with nothing to set up. Not for streams: see below. | | `load_job` | Writes it as a Parquet file to `gcs_bucket`, runs a BigQuery load job on it, waits for the job, then deletes the file. | Streams, and large or [parallel](/sdks/python/parallel-streaming) loads. | BigQuery keeps rows from streaming inserts in a buffer for a while, and refuses to delete rows still in it. So with `insert_rows`, a reorg can't remove the rows it replaces. Use `load_job` for anything that streams. Load jobs are also free, unlike streaming inserts, but each batch waits for its job, which adds latency. ## Load modes The BigQuery loader only appends: `mode='append'`, the default, is the only [load mode](/sdks/python/loaders#load-modes) it supports. To replace a table's rows, delete them, or the table, in BigQuery first. ## Streaming and reorgs (Unstable) With `stream=True`, the loader keeps loading new rows as Amp extracts blocks (see [Streaming](/sdks/python/streaming)). Each row's `_amp_batch_id` records the block range it came from. When a [reorg](/get-started/concepts/reorgs) replaces blocks, the loader finds the batches of those blocks in its stream state, and deletes their rows with a `DELETE` statement, a thousand batches at a time. Stream with `loading_method='load_job'`, so the rows can be deleted. :::warning The SDK's reorg detection has known issues, which make a stream of finished blocks reload them in a loop. See [Known issues](/sdks/python/streaming#known-issues) before you stream into BigQuery. ::: ## Troubleshooting ### `loading_method must be one of {…}, got '…'` `loading_method` isn't `'insert_rows'` or `'load_job'`. Fix its spelling. ### `loading_method='load_job' requires gcs_bucket to be set` Set `gcs_bucket` to the bucket to stage files in. ### `Invalid BigQueryLoader configuration: … missing 1 required positional argument: 'dataset'` The connection's configuration lacks a required option, named at the end. Add `project` and `dataset`. ### `Invalid BigQueryLoader configuration: … got an unexpected keyword argument` The configuration has an option the loader doesn't know, named at the end of the message. Check its spelling against [Configuration](/sdks/python/loaders/bigquery#configuration). ### `Your default credentials were not found.` The loader found no credentials. Run `gcloud auth application-default login`, set `GOOGLE_APPLICATION_CREDENTIALS`, or set `credentials_json` (see [Prerequisites](/sdks/python/loaders/bigquery#prerequisites)). ### `File … was not found.` `GOOGLE_APPLICATION_CREDENTIALS` points to a file that doesn't exist. Check the path. ### `Unsupported mode LoadMode.OVERWRITE` The loader only appends. Remove `mode='overwrite'`, and delete the table's rows in BigQuery before the load if you want to replace them. ### `Existing BigQuery table '…' is missing required '_amp_batch_id' column` You're loading into a table the loader didn't create. Add the column in BigQuery (`ALTER TABLE amp.blocks ADD COLUMN _amp_batch_id STRING`), or load into a new table. ### `insert_rows errors (first 3): …` BigQuery rejected rows from a streaming insert. The errors name each row and why, often a value that doesn't fit its column's type. Fix the query, or the table's schema. ### `would affect rows in the streaming buffer` A reorg tried to delete rows that `insert_rows` loaded recently, which BigQuery still holds in its streaming buffer. Stream with `loading_method='load_job'` instead. ### `Load job errors: …` A load job failed. The errors say why, often a permission on the bucket or the dataset, or a value that doesn't fit its column.