Google BigQuery

Technical Preview

Overview

A client-mode connector for the BigQuery API v2 that streams MQTT telemetry into BigQuery tables and runs SQL from MQTT:

Catalog connector event	Endpoint	Notes
Streaming insert	`POST /bigquery/v2/projects/{p}/datasets/{d}/tables/{t}/insertAll`	`tabledata.insertAll`. Row-level failures come back as `HTTP 200` with `insertErrors`.
Run SQL	`POST /bigquery/v2/projects/{p}/queries`	`jobs.query`. Synchronous up to `timeoutMs`; returns `jobReference`, `schema`, `rows`, `jobComplete`.
Query results	`GET /bigquery/v2/projects/{p}/queries/{jobId}?location=…&maxResults=…`	`jobs.getQueryResults`. Poll when `jobComplete` was `false`, or page with more rows.
Job status	`GET /bigquery/v2/projects/{p}/jobs/{jobId}?location=…`	`jobs.get`: state, statistics, error result.
List datasets	`GET /bigquery/v2/projects/{p}/datasets`	Credential smoke.
Create dataset / table	`POST …/datasets`, `POST …/datasets/{d}/tables`	One-time setup: dataset location and table schema.
List tables / rows	`GET …/datasets/{d}/tables`, `GET …/tables/{t}/data?maxResults=…`	Read-back with `tabledata.list`.

Default BASE_ADDRESS is https://bigquery.googleapis.com. Replace YOUR_PROJECT_ID in the endpoints with your GCP project id. Topics use data/google-bigquery/<object>/<verb>.

Storage Write API is not included. The BigQuery Storage Write API (and Storage Read API) is gRPC/protobuf only, so a REST_API connector can’t call it. This template uses the legacy tabledata.insertAll streaming endpoint, which is plain HTTPS + JSON. Load jobs with media upload (/upload/bigquery/v2/…) and destructive deletes are also left out.

Auth

BigQuery needs an OAuth 2.0 access token in Authorization: Bearer …. The template reads it from the MQTT payload field access_token; keep the token in a Hub secret such as GET SECRET "GOOGLE_BIGQUERY_ACCESS_TOKEN" and never commit it.

Minting the token is out of band. Exchange a service-account key (signed JWT) for an access token with gcloud auth print-access-token, a Google auth SDK, or your own token service, then publish or refresh it. Access tokens expire after about 1 hour.
Scopes: https://www.googleapis.com/auth/bigquery (full), https://www.googleapis.com/auth/bigquery.insertdata (streaming insert only), or https://www.googleapis.com/auth/cloud-platform.
IAM: the service account needs BigQuery Data Editor on the dataset and BigQuery Job User on the project to run queries.

A 401 UNAUTHENTICATED response with Request had invalid authentication credentials means the token is missing, malformed, or expired.

Basic Syntax

DEFINE ROUTE GoogleBigqueryRestApi WITH TYPE REST_API
    ADD REST_API_CONFIG
        WITH BASE_ADDRESS "https://bigquery.googleapis.com"
        WITH ENABLE_CLIENT "true"
        WITH USE_SSL "true"
        WITH REQUEST_TIMEOUT '60'
        WITH DEFAULT_CONTENT_TYPE "application/json"
        # Auth (prefer Hub secrets / GET SECRET "GOOGLE_BIGQUERY_ACCESS_TOKEN"):
        #   OAuth access_token on each MQTT payload → Bearer header below (refresh before ~1 h expiry).
        #   Aliases: BIGQUERY_ACCESS_TOKEN, GOOGLE_ACCESS_TOKEN.
        # Smoke: GET /bigquery/v2/projects/{p}/datasets → 200; POST /bigquery/v2/projects/{p}/queries
        #   {"query":"SELECT 1","useLegacySql":false} → 200 + jobComplete true.
    ADD EVENT ListDatasets
        WITH SOURCE_TOPIC "data/google-bigquery/datasets/list"
        WITH DESTINATION_TOPIC "data/google-bigquery/datasets/list/response"
        WITH METHOD "GET"
        WITH ENDPOINT "/bigquery/v2/projects/YOUR_PROJECT_ID/datasets"
        WITH HEADERS
            ADD "Accept" "application/json"
            ADD "Authorization" "Bearer {value.json.access_token}"
    ADD EVENT CreateDataset
        WITH SOURCE_TOPIC "data/google-bigquery/dataset/create"
        WITH DESTINATION_TOPIC "data/google-bigquery/dataset/create/response"
        WITH METHOD "POST"
        WITH ENDPOINT "/bigquery/v2/projects/YOUR_PROJECT_ID/datasets"
        WITH HEADERS
            ADD "Content-Type" "application/json"
            ADD "Authorization" "Bearer {value.json.access_token}"
        WITH BODY "{value.json.body}"
    ADD EVENT ListTables
        WITH SOURCE_TOPIC "data/google-bigquery/tables/list"
        WITH DESTINATION_TOPIC "data/google-bigquery/tables/list/response"
        WITH METHOD "GET"
        WITH ENDPOINT "/bigquery/v2/projects/YOUR_PROJECT_ID/datasets/{value.json.datasetId}/tables"
        WITH HEADERS
            ADD "Accept" "application/json"
            ADD "Authorization" "Bearer {value.json.access_token}"
    ADD EVENT CreateTable
        WITH SOURCE_TOPIC "data/google-bigquery/table/create"
        WITH DESTINATION_TOPIC "data/google-bigquery/table/create/response"
        WITH METHOD "POST"
        WITH ENDPOINT "/bigquery/v2/projects/YOUR_PROJECT_ID/datasets/{value.json.datasetId}/tables"
        WITH HEADERS
            ADD "Content-Type" "application/json"
            ADD "Authorization" "Bearer {value.json.access_token}"
        WITH BODY "{value.json.body}"
    ADD EVENT InsertRows
        WITH SOURCE_TOPIC "data/google-bigquery/rows/insert"
        WITH DESTINATION_TOPIC "data/google-bigquery/rows/insert/response"
        WITH METHOD "POST"
        WITH ENDPOINT "/bigquery/v2/projects/YOUR_PROJECT_ID/datasets/{value.json.datasetId}/tables/{value.json.tableId}/insertAll"
        WITH HEADERS
            ADD "Content-Type" "application/json"
            ADD "Authorization" "Bearer {value.json.access_token}"
        WITH BODY "{value.json.body}"
    ADD EVENT ListRows
        WITH SOURCE_TOPIC "data/google-bigquery/rows/list"
        WITH DESTINATION_TOPIC "data/google-bigquery/rows/list/response"
        WITH METHOD "GET"
        WITH ENDPOINT "/bigquery/v2/projects/YOUR_PROJECT_ID/datasets/{value.json.datasetId}/tables/{value.json.tableId}/data?maxResults={value.json.maxResults}"
        WITH HEADERS
            ADD "Accept" "application/json"
            ADD "Authorization" "Bearer {value.json.access_token}"
    ADD EVENT RunQuery
        WITH SOURCE_TOPIC "data/google-bigquery/query/run"
        WITH DESTINATION_TOPIC "data/google-bigquery/query/run/response"
        WITH METHOD "POST"
        WITH ENDPOINT "/bigquery/v2/projects/YOUR_PROJECT_ID/queries"
        WITH HEADERS
            ADD "Content-Type" "application/json"
            ADD "Authorization" "Bearer {value.json.access_token}"
        WITH BODY "{value.json.body}"
    ADD EVENT GetQueryResults
        WITH SOURCE_TOPIC "data/google-bigquery/query/results"
        WITH DESTINATION_TOPIC "data/google-bigquery/query/results/response"
        WITH METHOD "GET"
        WITH ENDPOINT "/bigquery/v2/projects/YOUR_PROJECT_ID/queries/{value.json.jobId}?location={value.json.location}&maxResults={value.json.maxResults}"
        WITH HEADERS
            ADD "Accept" "application/json"
            ADD "Authorization" "Bearer {value.json.access_token}"
    ADD EVENT GetJob
        WITH SOURCE_TOPIC "data/google-bigquery/job/get"
        WITH DESTINATION_TOPIC "data/google-bigquery/job/get/response"
        WITH METHOD "GET"
        WITH ENDPOINT "/bigquery/v2/projects/YOUR_PROJECT_ID/jobs/{value.json.jobId}?location={value.json.location}"
        WITH HEADERS
            ADD "Accept" "application/json"
            ADD "Authorization" "Bearer {value.json.access_token}"

Request bodies come from {value.json.body}, not {payload}. The MQTT payload carries control fields (access_token, datasetId, tableId, jobId, location, maxResults) next to a body object, and only body is sent to Google. Google APIs reject unknown top-level fields with 400 Invalid JSON payload received.

Example MQTT payloads

Create a dataset (topic data/google-bigquery/dataset/create):

{
  "access_token": "<from GOOGLE_BIGQUERY_ACCESS_TOKEN>",
  "body": { "datasetReference": { "projectId": "YOUR_PROJECT_ID", "datasetId": "coreflux_iiot" }, "location": "EU" }
}

Create a telemetry table (topic data/google-bigquery/table/create):

{
  "access_token": "<from GOOGLE_BIGQUERY_ACCESS_TOKEN>",
  "datasetId": "coreflux_iiot",
  "body": {
    "tableReference": { "projectId": "YOUR_PROJECT_ID", "datasetId": "coreflux_iiot", "tableId": "telemetry" },
    "schema": { "fields": [
      { "name": "ts", "type": "TIMESTAMP" },
      { "name": "site", "type": "STRING" },
      { "name": "temperature", "type": "FLOAT" }
    ] }
  }
}

Stream rows (topic data/google-bigquery/rows/insert). insertId gives best-effort de-duplication on retries:

{
  "access_token": "<from GOOGLE_BIGQUERY_ACCESS_TOKEN>",
  "datasetId": "coreflux_iiot",
  "tableId": "telemetry",
  "body": {
    "skipInvalidRows": false,
    "rows": [
      { "insertId": "porto-1", "json": { "ts": "2026-10-04T21:55:00Z", "site": "porto", "temperature": 21.5 } }
    ]
  }
}

A clean insert returns {"kind":"bigquery#tableDataInsertAllResponse"}. Rejected rows return HTTP 200 with insertErrors (for example no such field: pressure.), so check the response body, not just the status. Run SQL (topic data/google-bigquery/query/run):

{
  "access_token": "<from GOOGLE_BIGQUERY_ACCESS_TOKEN>",
  "body": {
    "query": "SELECT site, AVG(temperature) AS avg_t FROM `YOUR_PROJECT_ID.coreflux_iiot.telemetry` GROUP BY site",
    "useLegacySql": false,
    "location": "EU",
    "timeoutMs": 10000
  }
}

Fetch results or job status (topics data/google-bigquery/query/results, data/google-bigquery/job/get):

{ "access_token": "<from GOOGLE_BIGQUERY_ACCESS_TOKEN>", "jobId": "<jobReference.jobId>", "location": "EU", "maxResults": 100 }

Always send location and maxResults for the GET events, since the template puts them in the query string. Use the dataset’s location (US, EU, europe-west1, …); jobs outside the US and EU multi-regions are not found without it.

Quotas and limits

insertAll: up to 10 MB per request; Google recommends about 500 rows per request. Streamed rows are queryable within seconds, but they can’t be changed with DML for a short window after insert.
jobs.query waits up to timeoutMs (default 10 s). If jobComplete is false, poll GetQueryResults with jobReference.jobId.
On-demand queries are billed by bytes scanned. Use partitioned tables and WHERE filters on high-rate telemetry. See BigQuery quotas.

Testing

Status: blocked on credentials. The factory box has no Google OAuth credentials, so the connector has not been smoke-tested against Google. Wire format was checked against a local goccy/bigquery-emulator v0.8.1, which does not check OAuth: the request shapes for all nine events returned 200 (datasets list and create, table create and list, insertAll of two rows, tabledata.list read-back, jobs.query SELECT 1 and an aggregate read-back, getQueryResults, jobs.get). Against bigquery.googleapis.com without a valid token, GET …/datasets returned 401 UNAUTHENTICATED, as expected. To clear the gate, run GET /bigquery/v2/projects/{p}/datasets (expect 200) and POST /bigquery/v2/projects/{p}/queries with {"query":"SELECT 1","useLegacySql":false} (expect 200, jobComplete: true) using a real access token, project id, and lab dataset.

Quick Start

Coreflux MQTT Broker

LoT Language

Coreflux HUB

VS Code - LoT Notebooks

Coreflux Portal

Using AI with Coreflux

Product Updates

Overview

Auth

Basic Syntax

Example MQTT payloads

Quotas and limits

Testing

​Overview

​Auth

​Basic Syntax

​Example MQTT payloads

​Quotas and limits

​Testing

​Related

Overview

Auth

Basic Syntax

Example MQTT payloads

Quotas and limits

Testing

Related