# Ingest MQTT Data into BigQuery

[BigQuery](https://cloud.google.com/bigquery?hl=en) is a fully managed enterprise data warehouse for large-scale SQL-based analytics and reporting. EMQX Cloud can stream MQTT data into BigQuery through the rule engine and a BigQuery Sink, enabling real-time extraction, processing, and analysis of IoT data.

This page describes how to create a BigQuery data integration in EMQX Cloud. The example writes messages from the MQTT topic `test/a` into a BigQuery table.

## How It Works

The BigQuery integration uses the EMQX Cloud rule engine to select and transform MQTT messages, and then sends the rule output to BigQuery through a BigQuery Sink.

The data flow is:

```text
MQTT client -> EMQX Cloud -> Rule -> BigQuery Sink -> BigQuery table
```

1. MQTT clients publish telemetry or event data to a topic such as `test/a`.
2. A rule matches messages from the topic and selects the fields to write.
3. The BigQuery Sink writes the selected fields to the configured BigQuery dataset and table.
4. You can query the table in BigQuery to analyze the ingested MQTT data.

## Before You Start

### Prerequisites

Before you begin, make sure you are familiar with:

- [Data integration](./introduction.md)
- [Rules](./rules.md)
- Google Cloud service accounts, BigQuery datasets, and BigQuery tables

### Configure Network Access

The BigQuery connector connects to BigQuery over HTTPS. Configure the network according to your deployment type:

- For Dedicated Flex deployments, if your deployment can access Google Cloud services through the public network, enable the [NAT Gateway](../vas/nat-gateway.md) when required.
- For BYOC deployments, make sure the VPC where the deployment runs can access BigQuery. If public access is required, configure a NAT gateway in your cloud provider console.

### Create a Service Account Key in GCP

To allow EMQX Cloud to write data to BigQuery, create a service account in Google Cloud and generate a key in JSON format.

1. In your Google Cloud project, create a [service account](https://developers.google.com/identity/protocols/oauth2/service-account#creatinganaccount).

2. Grant the service account the permissions required to write to the target BigQuery dataset and table. For example, grant the **BigQuery Data Editor** role on the target dataset, or grant equivalent read and write permissions according to your security policy.

3. Open the service account details page, click the **Keys** tab, and create a new key in JSON format.

   ::: tip

   Keep the downloaded service account key secure. You will provide it when creating the BigQuery connector in EMQX Cloud.

   :::

### Create a Dataset and Table in BigQuery

Before configuring the BigQuery Sink in EMQX Cloud, create the target dataset and table in Google Cloud.

1. In the Google Cloud console, go to **BigQuery** -> **Studio**.

2. In the **Explorer** pane, create a dataset. For example, create a dataset named `emq_test_dataset`.

3. In the dataset, create a table. For example, create a table named `bigquery_integration_test`.

4. Define the table schema. For this tutorial, use the following schema:

   ```text
   clientid:string,payload:string,topic:string,publish_received_at:INTEGER
   ```

5. Verify that the service account you created has write permissions for the target table.

6. Optional: Run the following query to check that the table is accessible. Replace the project, dataset, and table names with your own values.

   ```sql
   SELECT * FROM `my_project.emq_test_dataset.bigquery_integration_test` LIMIT 1000
   ```

## Create a BigQuery Connector

Before creating a rule, create a BigQuery connector to connect EMQX Cloud to BigQuery.

1. In the EMQX Cloud Console, go to your deployment.

2. From the left navigation menu, click **Data Integration**.

3. If this is your first connector, select **BigQuery** under the **Data Persistence** category. If connectors already exist, click **New Connector**, and then select **BigQuery**.

4. On the **New Connector** page, configure the following fields:

   - **Connector Name**: Use the automatically generated name, or enter a name.
   - **GCP Service Account Credentials**: Paste the complete JSON content of the service account key created in [Create a Service Account Key in GCP](#create-a-service-account-key-in-gcp), or click **Select file** to import the JSON file.
   - Use the default values for other settings, or configure them according to your business needs.

5. Click **Test** to verify connectivity. If the BigQuery service is accessible and the service account credentials are valid, a success prompt is returned.

6. Click **New** to complete the connector setup. You can now create a rule and add a BigQuery Sink action.

## Create a Rule

Create a rule to select the MQTT message fields that will be written to BigQuery.

1. In the **Rules** section, click **New Rule**, or use the **Actions** icon next to your connector.

2. In the **SQL Editor**, enter the following SQL:

   ```sql
   SELECT
     clientid,
     topic,
     base64_encode(payload) AS payload,
     timestamp/1000 AS publish_received_at
   FROM
     "test/a"
   ```

   This rule listens for messages sent to the `test/a` topic and selects only the fields that match the BigQuery table schema.

   ::: tip

   BigQuery does not accept unknown fields. If you customize the SQL or table schema, make sure the selected field names match the columns in the target BigQuery table.

   :::

   ::: tip

   If you are a beginner user, click **Try It Out** to learn and test the SQL rule.

   :::

3. Click **Next** to append an action to the rule.

## Add a BigQuery Sink

On the **New Action (Sink)** page, configure a BigQuery Sink to write the rule output to BigQuery.

1. Configure the action:

   - **Connector**: Select the BigQuery connector you created.
   - **Action Type**: The value is **BigQuery**.
   - **Action Name**: Use the automatically generated name, or enter a name.
   - **Dataset**: Enter the BigQuery dataset name, for example, `emq_test_dataset`.
   - **Table**: Enter the BigQuery table name, for example, `bigquery_integration_test`.

2. Optional: Configure **Fallback Actions** if you want to improve reliability when message delivery fails. Fallback actions are triggered if the primary BigQuery Sink fails to process a message.

3. Keep **Advanced Settings** at the default values unless you need to tune connection or buffering behavior. For more information, see [Advanced Settings](#advanced-settings).

4. Click **Confirm** to create the rule and action.

5. In the **Successful new rule** pop-up, click **Back to Rules** to complete the rule creation.

## Test the Rule

Use MQTTX or another MQTT client to publish a test message to the `test/a` topic.

1. Publish the following message to EMQX Cloud:

   ```bash
   mqttx pub -i c_emqx -t test/a -m '{ "msg": "hello" }'
   ```

2. In the EMQX Cloud Console, go to the rule list and click the rule ID. Check the rule and action statistics. There should be one new incoming message and one new outgoing message for the BigQuery Sink.

3. In the Google Cloud console, go to **BigQuery** -> **Studio**, open the target table, and run the following query. Replace the project, dataset, and table names with your own values.

   ```sql
   SELECT *
   FROM `my_project.emq_test_dataset.bigquery_integration_test`
   ORDER BY publish_received_at DESC
   LIMIT 10
   ```

   You should see the message written to the target table.

   ![BigQuery query result](./_assets/bigquery_query_result.png)

If the query returns the test message, the integration is working correctly:

```text
MQTT -> Rule -> BigQuery Sink -> BigQuery table
```

## Advanced Settings

When configuring the BigQuery Sink, you can expand **Advanced Settings** to adjust the following parameters based on your needs.

| Field Name | Description | Default Value |
| ---------- | ----------- | ------------- |
| **Buffer Pool Size** | Specifies the number of buffer worker processes allocated to manage data flow between EMQX Cloud and BigQuery. These workers temporarily store and process data before sending it to BigQuery. | `16` |
| **Request TTL** | Specifies the maximum duration that a request remains valid after entering the buffer. If the request stays in the buffer longer than this value, or is sent but does not receive a timely response from BigQuery, the request expires. | `45` seconds |
| **Health Check Interval** | Specifies the time interval for the Sink to perform automatic health checks on its connection with BigQuery. | `15` seconds |
| **Health Check Interval Jitter** | Adds a random delay on top of the base health check interval to reduce the chance that multiple nodes initiate health checks at the same time. | `0` millisecond |
| **Health Check Timeout** | Specifies the timeout duration for connector health checks. | `5` seconds |
| **Max Buffer Queue Size** | Specifies the maximum number of bytes that can be buffered by each buffer worker process in the BigQuery Sink. | `16 MB`, `32 MB`, or `64 MB` (varies by deployment connection capacity) |
| **Query Mode** | Allows you to choose between synchronous and asynchronous request modes. In asynchronous mode, writing to BigQuery does not block the MQTT message publishing process. | `Async` |
| **Batch Size** | Specifies the maximum number of records sent to BigQuery in a single batch. If set to `1`, records are sent individually. | `1000` |
| **Inflight Window** | Specifies the maximum number of in-flight requests that can exist simultaneously during communication with BigQuery. When **Query Mode** is `Async`, set this value to `1` if strict sequential processing is required. | `100` |