# Amazon S3へのMQTTデータ取り込み [Amazon S3](https://aws.amazon.com/s3/)は、高い信頼性、安定性、セキュリティを備えたインターネットベースのストレージサービスであり、迅速なデプロイと使いやすさが特徴です。EMQXはMQTTメッセージをAmazon S3バケットに効率的に保存でき、柔軟なIoTデータストレージ機能を実現します。 本ページでは、EMQXとAmazon S3間のデータ統合について詳しく紹介し、ルールおよびSinkの作成方法について実践的なガイドを提供します。 :::tip EMQXはAmazon S3プロトコルに対応した他のストレージサービスとも互換性があります。例えば： - [MinIO](https://min.io/)：高性能で分散型のオブジェクトストレージシステム。Amazon S3 API互換のオープンソースオブジェクトストレージサーバーで、プライベートクラウド構築に適しています。 - [Google Cloud Storage](https://cloud.google.com/storage)：Google Cloudの統合オブジェクトストレージで、大量データの保存に対応。Amazon S3互換のインターフェースを提供します。 ビジネスニーズやシナリオに応じて適切なストレージサービスを選択できます。 ::: ## 動作概要 EMQXのAmazon S3データ統合はすぐに使える機能で、複雑なビジネス開発にも簡単に設定可能です。典型的なIoTアプリケーションでは、EMQXがデバイス接続とメッセージ伝送を担うIoTプラットフォームとして機能し、Amazon S3がメッセージデータの保存を担当するデータストレージプラットフォームとして利用されます。 ![emqx-integration-s3](./assets/emqx-integration-s3.jpg) EMQXはルールエンジンとSinkを利用してデバイスのイベントやデータをAmazon S3に転送します。アプリケーションはAmazon S3からデータを読み取り、さらなるデータ活用が可能です。具体的なワークフローは以下の通りです： 1. **デバイスのEMQX接続**：IoTデバイスはMQTTプロトコルで正常に接続するとオンラインイベントを発生させます。このイベントにはデバイスID、送信元IPアドレスなどのプロパティ情報が含まれます。 2. **デバイスメッセージのパブリッシュと受信**：デバイスは特定のトピックを通じてテレメトリやステータスデータをパブリッシュします。EMQXはメッセージを受信し、ルールエンジン内で比較処理を行います。 3. **ルールエンジンによるメッセージ処理**：組み込みのルールエンジンはトピックマッチングに基づき特定ソースのメッセージやイベントを処理します。該当するルールにマッチしたメッセージやイベントは、データ形式変換、特定情報のフィルタリング、コンテキスト情報の付加などの処理が行われます。 4. **Amazon S3への書き込み**：ルールはメッセージをS3に書き込むアクションをトリガーします。Amazon S3 Sinkを利用して処理結果からデータを抽出しS3に送信します。メッセージはテキストまたはバイナリ形式で保存可能で、複数行の構造化データはCSV、JSON Lines、Parquetファイルなどにまとめて保存できます。これはメッセージ内容やSinkの設定に依存します。 イベントやメッセージデータがAmazon S3に書き込まれた後は、Amazon S3に接続してデータを読み取り、以下のような柔軟なアプリケーション開発が可能です： - データアーカイブ：デバイスメッセージをオブジェクトとしてAmazon S3に長期保存し、コンプライアンス要件やビジネスニーズに対応。 - データ分析：S3から分析サービス（例：Snowflake）にデータを取り込み、予知保全やデバイス効率評価などの分析を実施。 ## 特長と利点 EMQXのAmazon S3データ統合を利用することで、以下の特長とメリットが得られます： - **メッセージ変換**：メッセージはEMQXルール内で大規模な処理・変換が可能で、Amazon S3への保存や利用に適した形に整形できます。 - **柔軟なデータ操作**：S3 Sinkを使うことで、特定フィールドのデータをAmazon S3バケットに簡単に書き込み可能。バケット名やオブジェクトキーを動的に設定でき、柔軟なデータ保存が実現します。 - **統合されたビジネスプロセス**：S3 SinkによりデバイスデータをAmazon S3の豊富なエコシステムアプリケーションと連携でき、データ分析やアーカイブなど多様なビジネスシナリオを実現します。 - **低コストの長期保存**：データベースと比較して、Amazon S3は高可用性・高信頼性かつコスト効率の良いオブジェクトストレージサービスであり、長期保存に適しています。 これらの特長により、効率的で信頼性が高くスケーラブルなIoTアプリケーションを構築し、ビジネスの意思決定や最適化に役立てることができます。 ## はじめる前に 本節では、EMQXでAmazon S3 Sinkを作成する前の準備事項を紹介します。 ### 前提条件 作業を進める前に、以下の内容を理解していることを確認してください。 #### EMQXの概念： - [ルールエンジン](./rules.md)：MQTTメッセージからデータを抽出・変換するロジックを定義する方法を理解する。 - [データ統合](./data-bridges.md)：EMQXのコネクターとSinkの概念を理解する。 #### AWSの概念： AWS S3が初めての場合は、以下の概念を確認してください： - **EC2**：AWSの仮想マシンサービス（コンピュートインスタンス）。 - **IAM**：AWSのIdentity and Access Management。インスタンスロールは、そのインスタンス上で動作するプログラムに一時的な認証情報を発行可能。 - **IMDSv2**：EC2のInstance Metadata Service v2。メタデータや一時認証情報を取得するためのトークンベースでより安全なサービス。 ### S3バケットの準備 EMQXはAmazon S3およびその他のS3互換ストレージサービスをサポートしています。AWSクラウドサービスを利用するか、DockerでMinIOインスタンスをデプロイできます。 :::: tabs ::: tab Amazon S3 1. [AWS S3コンソール](https://console.amazonaws.cn/s3/home)で、**Create bucket**ボタンをクリックします。バケット名やリージョンなどの必要情報を入力し、S3バケットを作成します。詳細は[AWSドキュメント](https://docs.amazonaws.cn/AmazonS3/latest/userguide/creating-bucket.html)を参照してください。 2. バケットの権限を設定します。バケット作成後、対象バケットを選択し、**Permissions**タブをクリックします。必要に応じてバケットをパブリック読み取り/書き込み、プライベートなどに設定可能です。 3. アクセスキーを取得します。 - **手動設定**：AWSコンソールで**IAM**サービスを検索・選択し、S3用の新規ユーザーを作成してAccess Key IDとSecret Access Keyを取得します。詳細は[AWSガイド：アクセスキーの管理](https://docs.aws.amazon.com/IAM/latest/UserGuide/id_credentials_access-keys.html)を参照してください。 - **自動取得（EC2のみ）**：EMQXが**AWS EC2**上で動作している場合は、十分な権限を持つ[**IAMロール**](https://docs.aws.amazon.com/IAM/latest/UserGuide/id_roles_use_switch-role-ec2.html)をアタッチします。EMQXは[**IMDSv2** API](https://docs.aws.amazon.com/AWSEC2/latest/UserGuide/instance-metadata-security-credentials.html)経由でインスタンスメタデータから一時認証情報を自動取得可能です。 Amazon S3バケットの作成と設定が完了したら、EMQXでAmazon S3 Sinkの作成準備が整います。 ::: ::: tab MinIO 1. DockerでMinIOをインストールして起動します： ```bash docker run \ -p 9000:9000 \ -p 9001:9001 \ --name minio \ -e "MINIO_ROOT_USER=admin" \ -e "MINIO_ROOT_PASSWORD=MyMinIOPassword" \ minio/minio:RELEASE.2024-02-17T01-15-57Z.fips \ server /data --console-address ":9001" ``` ポート`9000`はS3 API用、ポート`9001`はMinIO管理インターフェース用です。 起動後、ブラウザで`http://localhost:9001`にアクセスし、ログイン情報`admin`と`MyMinIOPassword`でMinIOコンソールに入れます。 2. バケットを作成します。MinIOコンソールで**Administrator** -> **Buckets**に移動し、ストレージバケット管理画面を開きます。右上の**Create Bucket +**ボタンをクリックし、`iot-data`と入力して**Create Bucket**をクリックしバケット作成を完了します。 3. アクセスキーを作成します。MinIOコンソールで**User** -> **Access Keys**に移動し、右上の**Create access key +**ボタンをクリックします。Access KeyとSecret Keyを入力し、**Create**をクリックしてアクセスキー作成を完了します。 MinIOのインストールと設定が完了したら、EMQXでAmazon S3 Sinkの作成準備が整います。 ::: :::: ## コネクターの作成 S3 Sinkを追加する前に、対応するコネクターを作成する必要があります。 1. ダッシュボードの**Integration** -> **Connector**ページに移動します。 2. 右上の**Create**ボタンをクリックします。 3. コネクタータイプとして**Amazon S3**を選択し、**Next**をクリックします。 4. コネクター名を入力します。名前は英数字で始まり、英数字、ハイフン、アンダースコアを含めることができます。ここでは例として`my-s3`と入力します。 5. 接続情報を入力します。 - Amazon S3バケットを使用する場合は、以下を入力します： - **Host**：リージョンにより異なり、`s3.{region}.amazonaws.com`形式です。 - **Port**：`443`を入力します。 - **Access Key ID**と**Secret Access Key**： - AWSで作成したアクセスキーを入力するか、 - EC2上でIAMロールをアタッチしている場合は空欄のままにします。 詳細は[Prepare an S3 Bucket](#prepare-an-s3-bucket)の「Amazon S3」タブを参照してください。 - MinIOを使用する場合は、以下を入力します： - **Host**：`127.0.0.1`を入力。リモートでMinIOを実行している場合は実際のホストアドレスを入力します。 - **Port**：`9000`を入力します。 - **Access Key ID**と**Secret Access Key**：MinIOで作成したアクセスキーを入力します。 6. 残りの設定はデフォルト値を使用します。 7. **Create**をクリックする前に、**Test Connectivity**をクリックしてコネクターがS3サービスに接続できるかテスト可能です。 8. 画面下部の**Create**ボタンをクリックしてコネクター作成を完了します。 これでコネクターの作成が完了し、次にルールとSinkを作成してS3に書き込むデータを指定します。 ## Amazon S3 Sink付きルールの作成 ここでは、EMQXでソースMQTTトピック`t/#`からメッセージを処理し、処理結果を設定済みのSinkを通じてS3の`iot-data`バケットに書き込むルールの作成方法を示します。 1. ダッシュボードの**Integration** -> **Rules**ページに移動します。 2. 右上の**Create**ボタンをクリックします。 3. ルールIDに`my_rule`を入力し、SQLエディターに以下のルールSQLを入力します： ```sql SELECT * FROM "t/#" ``` ::: tip SQLが初めての場合は、**SQL Examples**や**Enable Debug**をクリックしてルールSQLの学習やテストが可能です。 ::: 4. アクションを追加し、**Action Type**ドロップダウンから`Amazon S3`を選択します。アクションドロップダウンはデフォルトの`create action`のままにするか、既存のAmazon S3アクションを選択します。ここでは新しいSinkを作成してルールに追加します。 5. Sinkの名前と説明を入力します。 6. 先ほど作成した`my-s3`コネクターをコネクタードロップダウンから選択します。ドロップダウン横の作成ボタンをクリックしてポップアップで新規コネクターを素早く作成することも可能です。必要な設定パラメータは[Create a Connector](#create-a-connector)を参照してください。 7. **Bucket**に`iot-data`を入力します。このフィールドは`${var}`形式のプレースホルダーもサポートしますが、事前にS3に対応するバケットが作成されている必要があります。 8. 必要に応じて**ACL**を選択し、アップロードするオブジェクトのアクセス権限を指定します。 9. **Upload Method**を選択します。2つの方法の違いは以下の通りです： - **Direct Upload**：ルールがトリガーされるたびに、設定済みのオブジェクトキーと内容に従ってデータを直接S3にアップロードします。バイナリや大きなテキストデータの保存に適していますが、多数のファイルが生成される可能性があります。 - **Aggregated Upload**：複数のルールトリガー結果を1つのファイル（例：CSVファイル）にまとめてS3にアップロードします。構造化データの保存に適し、ファイル数を減らし書き込み効率を向上させます。 選択した方法に応じて設定パラメータが異なります。以下のタブから該当する方法の設定を行ってください。 :::: tabs type ::: tab Direct Upload Direct Uploadでは以下の項目を設定します： - **Object Key**：バケット内のオブジェクトの保存場所を定義します。`${var}`形式のプレースホルダーをサポートし、`/`でディレクトリ指定も可能です。管理や識別のためにオブジェクトのサフィックスも設定します。ここでは`msgs/${clientid}_${timestamp}.json`と入力します。`${clientid}`はクライアントID、`${timestamp}`はメッセージのタイムスタンプで、各デバイスのメッセージを異なるオブジェクトに書き込みます。 - **Object Content**：デフォルトはすべてのフィールドを含むJSONテキスト形式です。`${var}`形式のプレースホルダーをサポートします。ここでは`${payload}`を入力し、メッセージ本文をオブジェクト内容として使用します。オブジェクトの保存形式はメッセージ本文の形式に依存し、圧縮ファイルや画像、その他バイナリ形式もサポートします。 ::: ::: tab Aggregate Upload Aggregate Uploadでは以下のパラメータを設定します： - **Object Key**：オブジェクトの保存パスを指定します。以下の変数が使用可能です： - **`${action}`**：アクション名（必須）。 - **`${node}`**：アップロードを実行するEMQXノード名（必須）。 - **`${datetime.{format}}`**：集約開始日時。`{format}`でフォーマット指定（必須）： - **`${datetime.rfc3339utc}`**：UTC形式のRFC3339日時。 - **`${datetime.rfc3339}`**：ローカルタイムゾーン形式のRFC3339日時。 - **`${datetime.unix}`**：Unixタイムスタンプ。 - **`${datetime_until.{format}}`**：集約終了日時。フォーマットは上記と同様。 - **`${sequence}`**：同一時間間隔内の集約アップロードの連番（必須）。 必須マークのあるプレースホルダーがテンプレートに含まれていない場合、これらは自動的にS3オブジェクトキーのパスサフィックスとして追加され、重複を避けます。その他のプレースホルダーは無効とみなされます。 - **Aggregation Type**：S3に保存するバッチMQTTメッセージのデータファイル形式を定義します。サポート値： - `CSV`：カンマ区切りのCSV形式で書き込み。 - `JSON Lines`：[JSON Lines](https://jsonlines.org/)形式で書き込み。 - `parquet`：[Apache Parquet](https://parquet.apache.org/)形式で書き込み。列指向で大規模分析に最適化された形式です。 > スキーマ定義、圧縮、行グループ設定など詳細は[Parquet Format Options](#parquet-format-options)を参照してください。 - **Column Order**（`CSV`用）：ルール結果の列順をドロップダウンで調整可能。生成されるCSVファイルは選択列で先にソートされ、未選択列はアルファベット順で続きます。 - **Max Records**：最大レコード数に達すると1ファイルの集約を完了してアップロードし、時間間隔をリセットします。 - **Time Interval**：時間間隔に達すると最大レコード数に達していなくても1ファイルの集約を完了してアップロードし、最大レコード数をリセットします。 ::: :::: 11. **フォールバックアクション（任意）**：メッセージ配信失敗時の信頼性向上のため、1つ以上のフォールバックアクションを定義可能です。プライマリSinkがメッセージ処理に失敗した場合にこれらがトリガーされます。詳細は[Fallback Actions](./data-bridges.md#fallback-actions)を参照してください。 12. **Advanced Settings**を展開し、必要に応じて高度な設定オプションを構成します（任意）。詳細は[Advanced Settings](#advanced-settings)を参照してください。 13. 残りの設定はデフォルト値を使用し、**Create**ボタンをクリックしてSink作成を完了します。作成成功後はルール作成画面に戻り、新規Sinkがルールアクションに追加されます。 14. ルール作成画面で**Create**ボタンをクリックし、ルール作成全体を完了します。 これでルールの作成が完了し、**Rules**ページで新規ルールを確認でき、**Actions (Sink)**タブで新規S3 Sinkを確認できます。 また、**Integration** -> **Flow Designer**をクリックするとトポロジーを視覚的に確認できます。トポロジーはトピック`t/#`下のメッセージがルール`my_rule`で解析されS3に書き込まれる流れを示します。 ### Parquetフォーマットオプション **Aggregation Type**が`parquet`に設定されている場合、EMQXは集約されたルール結果をApache Parquet形式で保存します。Parquetは列指向で圧縮されたファイル形式で、分析処理に最適化されています。 本節ではParquet出力フォーマットの全設定オプションを説明します。 #### Parquetスキーマ（Avro） このオプションはMQTTメッセージのフィールドをParquetファイルの列にマッピングする方法を定義します。EMQXはApache Avroスキーマ仕様を用いてParquetデータの構造を記述します。 以下のいずれかを選択可能です： - **スキーマレジストリに存在するAvroスキーマ**：EMQXの[スキーマレジストリ](./schema-registry.md)で管理されている既存の[Avroスキーマ](./schema-registry-example-avro.md)を使用します。 この場合、シリアライズに使用するスキーマを識別するために**Schema Name**の指定が必要です。 ::: tip スキーマを中央管理し、複数システムで一貫したスキーマ進化を行いたい場合にこのオプションを使用してください。 ::: - **Avroスキーマを直接定義**：EMQX内でスキーマJSON構造を**Schema Definition**欄に直接入力して定義します。 例： ```json { "type": "record", "name": "MessageRecord", "fields": [ {"name": "clientid", "type": "string"}, {"name": "timestamp", "type": "long"}, {"name": "payload", "type": "string"} ] } ``` ::: tip フィールド名やデータ型はルールSQLが返すフィールドと一致させてください。不正または不足があるとParquet書き込み時にシリアライズエラーが発生する可能性があります。 ::: #### Parquetデフォルト圧縮 このオプションはParquetの各行グループ内のデータページに適用する圧縮アルゴリズムを指定します。圧縮によりストレージ容量を削減し、データクエリ時のI/O効率が向上します。 サポート値： | 値 | 説明 | | ------------------- | ------------------------------------------------------------ | | `snappy`（デフォルト） | 高速な圧縮・解凍と良好な圧縮率のバランスを持つ。ほとんどの場合に推奨。 | | `zstd` | 高圧縮率で中程度のCPU使用。大規模分析データや長期保存に最適。 | | `None` | 圧縮を無効化。デバッグ時や圧縮不要な場合に適する。 | #### Parquet最大行グループバイト数 Parquetの行グループは読み書きの基本単位であり、このオプションは1行グループの最大サイズ（バイト）を指定します。バッファ内のデータサイズがこの閾値を超えると、EMQXは現在の行グループをフラッシュし新規行グループを開始します。 - **デフォルト値**：`128 MB` ガイドライン： - **値を大きくする**とAthenaやSparkなど分析クエリの読み取り性能が向上します。 - **値を小さくする**と書き込み時のメモリ使用量を抑えられ、小規模データセットに適します。 ::: tip Parquetリーダーは行グループ単位でデータを読み込みます。大きな行グループはメタデータのオーバーヘッドを減らし分析クエリ性能を向上させます。 ::: ## ルールのテスト ここでは、Direct Upload方式で設定したルールのテスト方法を示します。 MQTTXを使い、トピック`t/1`にメッセージをパブリッシュします： ```bash mqttx pub -i emqx_c -t t/1 -m '{ "msg": "hello S3" }' ``` 数件メッセージを送信した後、MinIOコンソールまたはAmazon S3コンソールにアクセスして結果を確認します。 :::: tabs ::: tab Amazon S3コンソール AWSマネジメントコンソールにログインし、Amazon S3コンソールを開きます： バケット一覧から`ito-data`バケットを選択してバケット内に入ります。オブジェクト一覧に先ほどパブリッシュしたメッセージが`msg`オブジェクトとして正常に書き込まれているのが確認できます。オブジェクト横のチェックボックスを選択し、**Download**を選んでローカルにダウンロードし内容を確認可能です。 ::: ::: tab MinIOコンソール `iot-data`バケットを開きます。パブリッシュしたメッセージがMinIOの`msgs`ディレクトリに正常に書き込まれているのが確認できます： ![EMQX S3書き込み結果](./assets/emqx-integration-s3-test-result.png) ::: :::: ## 高度な設定 本節ではS3 Sinkの高度な設定オプションについて説明します。ダッシュボードでSinkを設定する際、**Advanced Settings**を展開し、ニーズに応じて以下のパラメータを調整可能です。 | フィールド名 | 説明 | デフォルト値 | | --------------------------------- | ------------------------------------------------------------ | --------------- | | **Buffer Pool Size** | EMQXとS3間のデータフローを管理するバッファワーカープロセスの数を指定します。これらのワーカーはデータを一時的に保存・処理し、ターゲットサービスへの送信を最適化しスムーズなデータ伝送を保証します。 | `16` | | **Request TTL** | バッファに入ったリクエストが有効とみなされる最大時間（秒）を指定します。リクエストがこのTTLを超えてバッファに滞留するか、送信後にS3からの応答やアックがタイムリーに得られない場合、リクエストは期限切れと見なされます。 | `45` | | **Health Check Interval** | SinkがS3との接続状態を自動的にヘルスチェックする間隔（秒）を指定します。 | `15`秒 | | **Health Check Interval Jitter** | 基本のヘルスチェック間隔に加える一様ランダム遅延時間（ミリ秒）です。複数ノードが同時にヘルスチェックを開始する確率を減らします。複数のアクションやソースが同一コネクターを共有する場合に有効です。 | `0`ミリ秒 | | **Health Check Timeout** | コネクターがS3との接続ヘルスチェックを行う際のタイムアウト時間（秒）を指定します。 | `60`秒 | | **Max Buffer Queue Size** | S3 Sinkの各バッファワーカーがバッファリング可能な最大バイト数を指定します。バッファワーカーはデータを一時保存し、S3への送信を効率化する仲介役です。システム性能やデータ伝送要件に応じて調整してください。 | `256` MB | | **Query Mode** | `synchronous`または`asynchronous`のリクエストモードを選択し、メッセージ伝送を最適化します。非同期モードではS3への書き込みがMQTTメッセージのパブリッシュをブロックしませんが、クライアントがS3到達前にメッセージを受信する可能性があります。 | `Asynchronous` | | **In-flight Window** | 「インフライトキューリクエスト」とは、開始済みだがまだ応答やアックを受け取っていないリクエストを指します。この設定はSinkとS3間の通信で同時に存在可能なインフライトリクエストの最大数を制御します。
`Request Mode`が`asynchronous`の場合に特に重要です。同一MQTTクライアントからのメッセージを厳密に順次処理する必要がある場合は、この値を`1`に設定してください。 | `100` | | **Min Part Size** | 集約完了後のパートアップロードの最小チャンクサイズです。アップロード対象データはこのサイズに達するまでメモリに蓄積されます。 | `5MB` | | **Max Part Size** | パートアップロードの最大チャンクサイズです。S3 Sinkはこのサイズを超えるパートのアップロードを試みません。 | `5GB` |