# データ統合の概要

EMQX Platformは、完全マネージド型のMQTTメッセージクラウドサービスとして、MQTTプロトコルを介してIoTデバイスを接続し、リアルタイムでメッセージを配信します。この基盤をもとに、データ統合はEMQX Platformの他のクラウドリソースとの接続機能を強化し、デバイスと他の業務システムとのシームレスな統合を可能にします。EMQX Platformのデータ統合は、明確かつ柔軟な「設定可能」なアーキテクチャソリューションを提供するだけでなく、開発プロセスを簡素化します。これによりユーザーの利用可能性が向上し、業務システムとEMQX Platform間の結合度を低減し、データ転送のためのより良いインフラを提供します。

![data_integration_intro](./_assets/integration_intro_01.png)

## 動作原理

デバイスやアプリケーションが接続を確立すると、MQTTブローカーはメッセージをルーティングします。到着したメッセージは、SQL文を用いてデータ操作を行う強力なコンポーネントであるルールエンジンによって処理されます。処理されたデータは「アクション」によってターゲットのサービスへ転送されます。アクションは「Sink」（サービスへデータを送信）と「Source」（サービスからデータを受信）の2種類に分類されます。現在、データ統合は「Sink」モードのみで動作し、様々なクラウドサービスへのデータのシームレスな統合を実現しています。

### [コネクター](./connectors.md)

コネクターはSink/Sourceの基盤となる接続チャネルであり、クラウドプラットフォームから購入したクラウドサービス製品に接続するために使用されます。クラウドサービス製品にはKafkaのようなメッセージキューサービスやRDSのようなストレージサービスがあります。

### [ルール](./rules.md)

ルールは「データの発生元」と「データのフィルタリングおよび処理方法」を記述します。ルールはSQLライクな文でカスタムデータを記述でき、SQLテストを用いてエクスポートされるデータのシミュレーションも可能です。ルールSQLの書き方については[ルールSQLリファレンス](./rule-sql-syntax.md)をご参照ください。

### [アクション](./rules.md#add-actions)

アクションは「処理済みデータの送信先」を決定します。ルールは1つ以上のアクションに対応でき、アクションは定義済みのコネクターを設定する必要があります。つまり、データの送信先を指定します。

#### Sink

Sinkはルールのアクションに追加されるデータ出力コンポーネントです。デバイスがイベントをトリガーしたり、メッセージがEMQXデプロイメントに到着すると、システムは対応するルールをマッチングして実行し、データをフィルタリングおよび処理します。ルールエンジンで処理されたデータは指定されたSinkに転送されます。Sinkでは、`${var}`や`${.var}`構文を使ってデータから変数を抽出し、SQL文やデータテンプレートを動的に生成するなど、データの取り扱い方法を設定できます。その後、対応する[コネクター](./connectors.md)を通じて外部データシステムにデータを送信し、メッセージの保存、データ更新、イベント通知などの操作を可能にします。

Sinkでサポートされる変数抽出構文は以下の通りです：

- `${var}`：ルールの出力結果から変数を抽出する構文です。例として`${topic}`があります。ネストされた変数を抽出したい場合はドット`.`を使い、`${payload.temp}`のように記述します。抽出したい変数が出力結果に含まれていない場合は、文字列`undefined`が返されます。
- `${.var}`：まずルールの出力結果から変数を抽出しようとし、存在しない場合は対応するイベントデータから抽出を試みます。例として`${.topic}`があります。こちらもドット`.`を使ったネスト変数の抽出をサポートし、`${.payload.temp}`のように記述できます。ルールの出力結果とイベントデータの両方に変数が存在しない場合は、文字列`undefined`が返されます。`${.}`を使うと、ルールの出力結果とイベントデータをマージしたすべての変数を抽出できます。

#### Source

Sourceはデータ入力コンポーネントであり、ルールの[データソース](../rule_engine/rule_engine_events.md)として機能し、ルールSQLを通じて選択されます。

SourceはMQTTやKafkaなどの外部データシステムからメッセージをサブスクライブまたはコンシュームします。コネクター経由で新しいメッセージが到着すると、ルールエンジンは対応するルールをマッチングして実行し、データをフィルタリングおよび処理します。処理されたデータは指定されたEMQXトピックにパブリッシュされ、クラウドコマンドの配信などの操作を可能にします。

## ワークフロー

以下はデータ統合を作成する基本的な手順です：

![data_integration_intro](./_assets/integration_intro_02.png)

1. コネクターを作成します。デプロイメントのデータ統合初期ページから接続したいサービスを選択し、コネクターを設定します。
2. デバイスから収集したデータを処理するルールを作成します。ルールはSQL文を使って任意の方法でデータを収集・処理できます。
3. ルールにアクションを追加します。ルールがトリガーされると、処理されたデータは設定されたコネクターを通じてクラウドサービスに転送されます。
4. 作成したデータ統合が正しく動作するかテストします。

## デプロイメントに必要なネットワーク設定

データ統合機能はデプロイメントの種類によって、データソースへのアクセスレベルやネットワーク構成が異なります。

**サーバーレスデプロイメント**

- データソースはパブリックネットワークアクセスのみをサポートします。そのため、データソース作成前にパブリックネットワークアクセスが可能であることと、セキュリティグループが開放されていることを確認してください。
- KafkaおよびHTTPサーバーコネクターのみサポートします。
- サーバーレスデプロイメントのデータ統合は従量課金制で、以下の通りです：

  EMQXサーバーレスはユーザーに月間最大100万回のルールアクション実行無料枠を提供します。これを超えると、追加の100万回のルールアクション実行ごとに0.25ドルの料金が発生します。

  最適なパフォーマンスと管理性を維持するため、EMQX Platformは各デプロイメント内のコネクター、ルール、アクションの作成に以下の制限を設けています：

  | カテゴリー                  | 最大許容数       |
  | --------------------------- | --------------- |
  | コネクター合計             | 2               |
  | ルール合計                 | 4               |
  | ルールあたりのアクション数 | 1               |

**専用デプロイメント**

- データソースへは内部ネットワーク経由でのアクセスを推奨します。そのため、作成前に[VPCピアリング](../deployments/vpc_peering.md)を設定し、セキュリティグループを開放してください。
- パブリックネットワーク経由でアクセスする場合は[NATゲートウェイ](../vas/nat-gateway.md)を有効にできます。

**BYOCデプロイメント**

:::tip 注意

BYOCデプロイメントではコンソールからのデータ統合設定はできません。設定には[EMQX Management Console](../deployments/byoc_management.md)をご利用ください。

:::

- ネットワークセキュリティとパフォーマンス向上のため、データソースへは内部ネットワーク経由でのアクセスを推奨します。作成前に、リソースが存在するVPCとBYOCデプロイメントが存在するパブリッククラウドコンソールのVPC間でピアリング接続を設定し、関連するセキュリティグループを開放してください。詳細は[Create VPC Peering Connections](../deployments/byoc_vpc_peering.md)をご参照ください。
- パブリックネットワーク経由でリソースにアクセスする場合は、BYOCデプロイメントが存在するVPCに対してパブリッククラウドコンソールでNATゲートウェイを設定してください。