# BigQuery に MQTT データを取り込む

[BigQuery](https://cloud.google.com/bigquery?hl=en) は、大規模な SQL ベースの分析およびレポーティング向けのフルマネージド型エンタープライズデータウェアハウスです。EMQX Cloud はルールエンジンと BigQuery Sink を通じて MQTT データを BigQuery にストリーミングでき、IoT データのリアルタイム抽出、処理、分析を可能にします。

このページでは、EMQX Cloud で BigQuery のデータ統合を作成する方法について説明します。例として、MQTT トピック `test/a` からのメッセージを BigQuery のテーブルに書き込みます。

## 動作の仕組み

BigQuery 統合は EMQX Cloud のルールエンジンを使用して MQTT メッセージを選択・変換し、ルールの出力を BigQuery Sink を介して BigQuery に送信します。

データフローは以下の通りです：

```text
MQTT クライアント -> EMQX Cloud -> ルール -> BigQuery Sink -> BigQuery テーブル
```

1. MQTT クライアントが `test/a` のようなトピックにテレメトリやイベントデータをパブリッシュする。
2. ルールがトピックからのメッセージにマッチし、書き込むフィールドを選択する。
3. BigQuery Sink が選択されたフィールドを設定された BigQuery のデータセットとテーブルに書き込む。
4. BigQuery でテーブルをクエリし、取り込んだ MQTT データを分析できる。

## はじめる前に

### 前提条件

開始する前に、以下を理解していることを確認してください：

- [データ統合](./introduction.md)
- [ルール](./rules.md)
- Google Cloud のサービスアカウント、BigQuery データセット、および BigQuery テーブル

### ネットワークアクセスの設定

BigQuery コネクターは HTTPS 経由で BigQuery に接続します。デプロイタイプに応じてネットワークを設定してください：

- Dedicated Flex デプロイメントの場合、デプロイメントがパブリックネットワーク経由で Google Cloud サービスにアクセス可能なら、必要に応じて [NAT ゲートウェイ](../vas/nat-gateway.md) を有効にしてください。
- BYOC デプロイメントの場合、デプロイメントが稼働する VPC が BigQuery にアクセスできることを確認してください。パブリックアクセスが必要な場合は、クラウドプロバイダーのコンソールで NAT ゲートウェイを設定してください。

### GCP でサービスアカウントキーを作成する

EMQX Cloud が BigQuery にデータを書き込めるように、Google Cloud でサービスアカウントを作成し、JSON 形式のキーを生成します。

1. Google Cloud プロジェクトで、[サービスアカウント](https://developers.google.com/identity/protocols/oauth2/service-account#creatinganaccount) を作成します。

2. サービスアカウントに、対象の BigQuery データセットおよびテーブルに書き込むために必要な権限を付与します。例として、対象データセットに対して **BigQuery Data Editor** ロールを付与するか、セキュリティポリシーに応じた同等の読み書き権限を付与してください。

3. サービスアカウントの詳細ページを開き、**Keys** タブで新しいキーを JSON 形式で作成します。

   ::: tip

   ダウンロードしたサービスアカウントキーは安全に保管してください。EMQX Cloud で BigQuery コネクターを作成する際に使用します。

   :::

### BigQuery でデータセットとテーブルを作成する

EMQX Cloud で BigQuery Sink を設定する前に、Google Cloud で対象のデータセットとテーブルを作成してください。

1. Google Cloud コンソールで **BigQuery** -> **Studio** に移動します。

2. **Explorer** ペインでデータセットを作成します。例として `emq_test_dataset` という名前のデータセットを作成します。

3. データセット内にテーブルを作成します。例として `bigquery_integration_test` という名前のテーブルを作成します。

4. テーブルのスキーマを定義します。このチュートリアルでは以下のスキーマを使用します：

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

5. 作成したサービスアカウントが対象テーブルへの書き込み権限を持っていることを確認してください。

6. 任意で、以下のクエリを実行してテーブルにアクセス可能か確認します。プロジェクト名、データセット名、テーブル名はご自身の環境に置き換えてください。

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

## BigQuery コネクターを作成する

ルールを作成する前に、EMQX Cloud と BigQuery を接続する BigQuery コネクターを作成します。

1. EMQX Cloud コンソールで、対象のデプロイメントに移動します。

2. 左側のナビゲーションメニューから **データ統合** をクリックします。

3. 初めてのコネクターの場合は、**データ永続化** カテゴリの中から **BigQuery** を選択します。既にコネクターが存在する場合は、**新規コネクター** をクリックし、**BigQuery** を選択します。

4. **新規コネクター** ページで以下の項目を設定します：

   - **コネクター名**：自動生成された名前を使うか、任意の名前を入力します。
   - **GCP サービスアカウント認証情報**：[GCP でサービスアカウントキーを作成する](#gcp-でサービスアカウントキーを作成する) で作成したサービスアカウントキーの JSON 全体を貼り付けるか、**ファイルを選択** をクリックして JSON ファイルをインポートします。
   - その他の設定はデフォルトのままか、ビジネスニーズに応じて設定してください。

5. **テスト** をクリックして接続を検証します。BigQuery サービスにアクセスでき、サービスアカウント認証情報が有効なら成功メッセージが表示されます。

6. **新規作成** をクリックしてコネクターの設定を完了します。これでルールを作成し、BigQuery Sink アクションを追加できます。

## ルールを作成する

BigQuery に書き込む MQTT メッセージのフィールドを選択するルールを作成します。

1. **ルール** セクションで **新規ルール** をクリックするか、コネクター横の **アクション** アイコンを使用します。

2. **SQL エディター** に以下の SQL を入力します：

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

   このルールは `test/a` トピックに送信されたメッセージを監視し、BigQuery テーブルのスキーマに合致するフィールドのみを選択します。

   ::: tip

   BigQuery は未知のフィールドを受け付けません。SQL やテーブルスキーマをカスタマイズする場合は、選択するフィールド名が対象の BigQuery テーブルのカラム名と一致していることを確認してください。

   :::

   ::: tip

   初心者の方は **Try It Out** をクリックして SQL ルールの学習とテストを行ってください。

   :::

3. **次へ** をクリックしてルールにアクションを追加します。

## BigQuery Sink を追加する

**新規アクション（Sink）** ページで、ルールの出力を BigQuery に書き込む BigQuery Sink を設定します。

1. アクションの設定：

   - **コネクター**：作成した BigQuery コネクターを選択します。
   - **アクションタイプ**：値は **BigQuery** です。
   - **アクション名**：自動生成された名前を使うか、任意の名前を入力します。
   - **データセット**：BigQuery のデータセット名を入力します。例：`emq_test_dataset`
   - **テーブル**：BigQuery のテーブル名を入力します。例：`bigquery_integration_test`

2. 任意で、メッセージ配信失敗時の信頼性向上のために **フォールバックアクション** を設定できます。フォールバックアクションは、プライマリの BigQuery Sink がメッセージ処理に失敗した場合にトリガーされます。

3. **詳細設定** は接続やバッファリングの挙動を調整する必要がなければデフォルトのままにしてください。詳細は [詳細設定](#詳細設定) を参照してください。

4. **確認** をクリックしてルールとアクションを作成します。

5. **新規ルール作成成功** ポップアップで **ルールに戻る** をクリックしてルール作成を完了します。

## ルールをテストする

MQTTX などの MQTT クライアントを使い、`test/a` トピックにテストメッセージをパブリッシュします。

1. 以下のメッセージを EMQX Cloud にパブリッシュします：

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

2. EMQX Cloud コンソールでルール一覧に移動し、ルール ID をクリックします。ルールとアクションの統計情報を確認してください。BigQuery Sink に対して新しい受信メッセージと送信メッセージがそれぞれ 1 件あるはずです。

3. Google Cloud コンソールで **BigQuery** -> **Studio** に移動し、対象テーブルを開いて以下のクエリを実行します。プロジェクト名、データセット名、テーブル名はご自身の環境に置き換えてください。

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

   対象テーブルにメッセージが書き込まれていることが確認できます。

   ![BigQuery クエリ結果](./_assets/bigquery_query_result.png)

クエリでテストメッセージが返されれば、統合は正常に機能しています：

```text
MQTT -> ルール -> BigQuery Sink -> BigQuery テーブル
```

## 詳細設定

BigQuery Sink の設定時に **詳細設定** を展開すると、ニーズに応じて以下のパラメーターを調整できます。

| フィールド名                 | 説明                                                                                         | デフォルト値           |
| ---------------------------- | -------------------------------------------------------------------------------------------- | ---------------------- |
| **バッファプールサイズ**      | EMQX Cloud と BigQuery 間のデータフローを管理するバッファワーカープロセスの数を指定します。これらのワーカーはデータを一時的に保存・処理し、BigQuery に送信します。 | `16`                   |
| **リクエスト TTL**            | リクエストがバッファに入ってから有効な最大時間を指定します。この時間を超えてバッファに残るか、送信後に BigQuery からタイムリーな応答が得られない場合、リクエストは期限切れになります。 | `45` 秒                |
| **ヘルスチェック間隔**        | Sink が BigQuery との接続状態を自動的にチェックする間隔を指定します。                             | `15` 秒                |
| **ヘルスチェック間隔ジッター** | 複数のノードが同時にヘルスチェックを開始する可能性を減らすため、基本間隔にランダムな遅延を追加します。           | `0` ミリ秒             |
| **ヘルスチェックタイムアウト** | コネクターのヘルスチェックのタイムアウト時間を指定します。                                     | `5` 秒                 |
| **最大バッファキューサイズ**  | BigQuery Sink の各バッファワーカーがバッファリングできる最大バイト数を指定します。                   | `16 MB`、`32 MB`、`64 MB`（デプロイ接続容量により異なる） |
| **クエリモード**              | 同期モードと非同期モードを選択できます。非同期モードでは、BigQuery への書き込みが MQTT メッセージのパブリッシュ処理をブロックしません。 | `Async`                |
| **バッチサイズ**              | 1 回のバッチで BigQuery に送信するレコードの最大数を指定します。`1` に設定するとレコードを個別送信します。 | `1000`                 |
| **インフライトウィンドウ**    | BigQuery との通信中に同時に存在できるインフライトリクエストの最大数を指定します。**クエリモード** が `Async` の場合、厳密な順次処理が必要ならこの値を `1` に設定してください。 | `100`                  |