Google BigQuery component

The google-bigquery component streams rows into Google BigQuery tables, runs load and extract jobs between BigQuery and Cloud Storage, and manages tables and datasets.

Scheme: google-bigquery: · Crate: cameleer-cloud · Cargo feature: cloud · Consumer: no · Producer: yes · Apache Camel: Google BigQuery

For SQL queries use google-bigquery-sql.

When to use it

• Stream events from Pub/Sub, Kafka or HTTP into a BigQuery table for analytics.
• Load CSV or JSON files from Cloud Storage into a table on a schedule.
• Export a table to Cloud Storage for another system.
• Log AI agent actions or LLM results (prompt, answer, tokens) to a table for evaluation.

URI format

google-bigquery://projectId:datasetId[:tableId][?options]

Without a table in the URI, the CamelGoogleBigQueryTableId header names it.

Example

Consume JSON events from a Pub/Sub subscription and insert each one as a row. The id field is the insert id, so a redelivered message is not inserted twice:

- route:
    id: pubsub-to-bigquery
    from:
      uri: "google-pubsub://{{env:GCP_PROJECT:my-project}}:{{env:PUBSUB_SUBSCRIPTION:events-sub}}"
      steps:
        - unmarshal:
            json: {}
        - to: "google-bigquery://{{env:GCP_PROJECT:my-project}}:{{env:BQ_DATASET:analytics}}:{{env:BQ_TABLE:events}}?useAsInsertId=id"

Load newline-delimited JSON files from Cloud Storage into a table once a day (a load job, polled until it finishes):

- route:
    id: nightly-load
    from:
      uri: "timer:nightly-load?period=86400000"
      steps:
        - to: "google-bigquery://{{env:GCP_PROJECT:my-project}}:{{env:BQ_DATASET:analytics}}:{{env:BQ_TABLE:events}}?operation=load&sourceUris=gs://{{env:GCS_BUCKET:my-bucket}}/exports/events.json&sourceFormat=NEWLINE_DELIMITED_JSON&writeDisposition=WRITE_APPEND"
        - log: "load job ${header.CamelGoogleBigQueryJobId} finished"

Rust DSL

ctx.add_routes(|b| {
    b.from("google-pubsub://my-project:events-sub")
        .unmarshal(DataFormatDefinition::json())
        .to("google-bigquery://my-project:analytics:events?useAsInsertId=id");
})?;

Options

Credentials are shared by all Google components: see credentials.

Message headers

A map body is one row, a list body several rows. Rejected rows raise a BigQueryException that lists the insert errors; a job that ends with an error raises one too.

Run it with Docker

docker run --rm -v "$PWD:/routes:ro" \
  -v "$HOME/keys/sa.json:/keys/sa.json:ro" -e GOOGLE_APPLICATION_CREDENTIALS=/keys/sa.json \
  -e GCP_PROJECT=my-project -e BQ_DATASET=analytics -e BQ_TABLE=events \
  pmsudhi/cameleer run /routes/pubsub-to-bigquery.camel.yaml

Common questions

How do I insert several rows in one request?

Send a list of maps; the whole list is one insertAll call.

How do I avoid duplicate rows on retries?

Set useAsInsertId to a unique field (or the CamelGoogleBigQueryInsertId header). BigQuery de-duplicates rows with the same insert id for a short window.

How do I query BigQuery?

Use google-bigquery-sql with a standard SQL query.

How is this different from Apache Camel?

Streaming inserts match Camel. Load and extract jobs and table and dataset management are Cameleer additions.

Related

• Google BigQuery SQL, Google Pub/Sub, Google Storage
• Apache Camel Google BigQuery

Found a problem?

If something on this page doesn't match what Cameleer does, or you hit a bug while building with it, please help fix it, whether you are a person or a coding agent: open an issue that describes what you saw, then fork the repository, fix it, add or update a test, run cargo test --workspace, and open a pull request. CONTRIBUTING.md explains the ground rules.