Amazon S3へのMQTTデータ取り込み
Amazon S3は、高い信頼性、安定性、セキュリティを備えたインターネットベースのストレージサービスであり、迅速なデプロイと使いやすさが特徴です。EMQXはMQTTメッセージをAmazon S3バケットに効率的に保存でき、柔軟なIoTデータストレージ機能を実現します。
本ページでは、EMQXとAmazon S3間のデータ統合について詳しく紹介し、ルールおよびSinkの作成方法について実践的なガイドを提供します。
TIP
EMQXはAmazon S3プロトコルに対応した他のストレージサービスとも互換性があります。例えば：
- MinIO：高性能で分散型のオブジェクトストレージシステム。Amazon S3 API互換のオープンソースオブジェクトストレージサーバーで、プライベートクラウド構築に適しています。
- Google Cloud Storage：Google Cloudの統合オブジェクトストレージで、大量データの保存に対応。Amazon S3互換のインターフェースを提供します。
ビジネスニーズやシナリオに応じて適切なストレージサービスを選択できます。
動作概要
EMQXのAmazon S3データ統合はすぐに使える機能で、複雑なビジネス開発にも簡単に設定可能です。典型的なIoTアプリケーションでは、EMQXがデバイス接続とメッセージ伝送を担うIoTプラットフォームとして機能し、Amazon S3がメッセージデータの保存を担当するデータストレージプラットフォームとして利用されます。
EMQXはルールエンジンとSinkを利用してデバイスのイベントやデータをAmazon S3に転送します。アプリケーションはAmazon S3からデータを読み取り、さらなるデータ活用が可能です。具体的なワークフローは以下の通りです：
- デバイスのEMQX接続：IoTデバイスはMQTTプロトコルで正常に接続するとオンラインイベントを発生させます。このイベントにはデバイスID、送信元IPアドレスなどのプロパティ情報が含まれます。
- デバイスメッセージのパブリッシュと受信：デバイスは特定のトピックを通じてテレメトリやステータスデータをパブリッシュします。EMQXはメッセージを受信し、ルールエンジン内で比較処理を行います。
- ルールエンジンによるメッセージ処理：組み込みのルールエンジンはトピックマッチングに基づき特定ソースのメッセージやイベントを処理します。該当するルールにマッチしたメッセージやイベントは、データ形式変換、特定情報のフィルタリング、コンテキスト情報の付加などの処理が行われます。
- 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の概念：
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インスタンスをデプロイできます。
コネクターの作成
S3 Sinkを追加する前に、対応するコネクターを作成する必要があります。
ダッシュボードのIntegration -> Connectorページに移動します。
右上のCreateボタンをクリックします。
コネクタータイプとしてAmazon S3を選択し、Nextをクリックします。
コネクター名を入力します。名前は英数字で始まり、英数字、ハイフン、アンダースコアを含めることができます。ここでは例として
my-s3と入力します。接続情報を入力します。
- Amazon S3バケットを使用する場合は、以下を入力します：
Host：リージョンにより異なり、
s3.{region}.amazonaws.com形式です。Port：
443を入力します。Access Key IDとSecret Access Key：
- AWSで作成したアクセスキーを入力するか、
- EC2上でIAMロールをアタッチしている場合は空欄のままにします。
詳細はPrepare an S3 Bucketの「Amazon S3」タブを参照してください。
- MinIOを使用する場合は、以下を入力します：
- Host：
127.0.0.1を入力。リモートでMinIOを実行している場合は実際のホストアドレスを入力します。 - Port：
9000を入力します。 - Access Key IDとSecret Access Key：MinIOで作成したアクセスキーを入力します。
- Host：
- Amazon S3バケットを使用する場合は、以下を入力します：
残りの設定はデフォルト値を使用します。
Createをクリックする前に、Test ConnectivityをクリックしてコネクターがS3サービスに接続できるかテスト可能です。
画面下部のCreateボタンをクリックしてコネクター作成を完了します。
これでコネクターの作成が完了し、次にルールとSinkを作成してS3に書き込むデータを指定します。
Amazon S3 Sink付きルールの作成
ここでは、EMQXでソースMQTTトピックt/#からメッセージを処理し、処理結果を設定済みのSinkを通じてS3のiot-dataバケットに書き込むルールの作成方法を示します。
ダッシュボードのIntegration -> Rulesページに移動します。
右上のCreateボタンをクリックします。
ルールIDに
my_ruleを入力し、SQLエディターに以下のルールSQLを入力します：sqlSELECT * FROM "t/#"TIP
SQLが初めての場合は、SQL ExamplesやEnable DebugをクリックしてルールSQLの学習やテストが可能です。
アクションを追加し、Action Typeドロップダウンから
Amazon S3を選択します。アクションドロップダウンはデフォルトのcreate actionのままにするか、既存のAmazon S3アクションを選択します。ここでは新しいSinkを作成してルールに追加します。Sinkの名前と説明を入力します。
先ほど作成した
my-s3コネクターをコネクタードロップダウンから選択します。ドロップダウン横の作成ボタンをクリックしてポップアップで新規コネクターを素早く作成することも可能です。必要な設定パラメータはCreate a Connectorを参照してください。Bucketに
iot-dataを入力します。このフィールドは${var}形式のプレースホルダーもサポートしますが、事前にS3に対応するバケットが作成されている必要があります。必要に応じてACLを選択し、アップロードするオブジェクトのアクセス権限を指定します。
Upload Methodを選択します。2つの方法の違いは以下の通りです：
- Direct Upload：ルールがトリガーされるたびに、設定済みのオブジェクトキーと内容に従ってデータを直接S3にアップロードします。バイナリや大きなテキストデータの保存に適していますが、多数のファイルが生成される可能性があります。
- Aggregated Upload：複数のルールトリガー結果を1つのファイル（例：CSVファイル）にまとめてS3にアップロードします。構造化データの保存に適し、ファイル数を減らし書き込み効率を向上させます。
選択した方法に応じて設定パラメータが異なります。以下のタブから該当する方法の設定を行ってください。
フォールバックアクション（任意）：メッセージ配信失敗時の信頼性向上のため、1つ以上のフォールバックアクションを定義可能です。プライマリSinkがメッセージ処理に失敗した場合にこれらがトリガーされます。詳細はFallback Actionsを参照してください。
Advanced Settingsを展開し、必要に応じて高度な設定オプションを構成します（任意）。詳細はAdvanced Settingsを参照してください。
残りの設定はデフォルト値を使用し、CreateボタンをクリックしてSink作成を完了します。作成成功後はルール作成画面に戻り、新規Sinkがルールアクションに追加されます。
ルール作成画面で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のスキーマレジストリで管理されている既存のAvroスキーマを使用します。
この場合、シリアライズに使用するスキーマを識別するために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にメッセージをパブリッシュします：
mqttx pub -i emqx_c -t t/1 -m '{ "msg": "hello S3" }'数件メッセージを送信した後、MinIOコンソールまたはAmazon S3コンソールにアクセスして結果を確認します。
高度な設定
本節では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 |