EMQX Cloud

日本語

简体中文
English

日本語

简体中文
English

Appearance

お問い合わせ
お問い合わせ

Azure Blob Storage に MQTT データを取り込む ​

Azure Blob Storage は、マイクロソフトが提供するクラウドベースのオブジェクトストレージソリューションで、大量の非構造化データの取り扱いに特化しています。非構造化データとは、特定のデータモデルやフォーマットに従わないデータタイプ（テキストファイルやバイナリデータなど）を指します。EMQX Cloud は MQTT メッセージを効率的に Blob Storage コンテナに保存でき、IoT データの保存に柔軟なソリューションを提供します。

本ページでは、EMQX Cloud と Azure Blob Storage 間のデータ統合について詳しく解説し、ルールおよび Sink の作成方法を実践的に案内します。

動作概要 ​

EMQX Cloud における Azure Blob Storage データ統合は、すぐに利用可能な機能であり、複雑なビジネス開発にも簡単に設定できます。典型的な IoT アプリケーションでは、EMQX Cloud がデバイス接続とメッセージ伝送を担う IoT プラットフォームとして機能し、Azure Blob Storage はメッセージデータの保存を担当するデータストレージプラットフォームとして利用されます。

EMQX Cloud Azure Blob Storage

EMQX Cloud はルールエンジンと Sink を活用してデバイスのイベントやデータを Azure Blob Storage に転送します。アプリケーションは Azure Blob Storage からデータを読み取り、さらなるデータ活用が可能です。具体的なワークフローは以下の通りです。

  1. デバイスの EMQX Cloud への接続：IoT デバイスは MQTT プロトコルで正常に接続するとオンラインイベントをトリガーします。このイベントにはデバイスID、送信元IPアドレスなどのプロパティ情報が含まれます。
  2. デバイスのメッセージパブリッシュと受信：デバイスは特定のトピックを通じてテレメトリやステータスデータをパブリッシュします。EMQX Cloud はこれらのメッセージを受信し、ルールエンジン内で照合します。
  3. ルールエンジンによるメッセージ処理：組み込みのルールエンジンはトピックマッチングに基づき特定のソースからのメッセージやイベントを処理します。対応するルールをマッチングし、データ形式の変換、特定情報のフィルタリング、コンテキスト情報によるメッセージの付加などを行います。
  4. Azure Blob Storage への書き込み：ルールはメッセージをストレージコンテナに書き込むアクションをトリガーします。Azure Blob Storage Sink を利用することで、処理結果からデータを抽出し Blob Storage に送信可能です。メッセージはテキストまたはバイナリ形式で保存でき、メッセージ内容や Sink の設定に応じて複数行の構造化データを単一の CSV ファイルにまとめることも可能です。

イベントやメッセージデータがストレージコンテナに書き込まれた後は、Azure Blob Storage に接続してデータを読み取り、以下のような柔軟なアプリケーション開発に活用できます。

  • データアーカイブ：デバイスメッセージを Azure Blob Storage のオブジェクトとして長期保存し、コンプライアンス要件やビジネスニーズに対応。
  • データ分析：ストレージコンテナからデータを分析サービス（例：Snowflake）に取り込み、予知保全やデバイス効率評価などのデータ分析サービスを実現。

特長と利点 ​

EMQX Cloud における Azure Blob Storage データ統合を利用することで、以下の特長と利点が得られます。

  • メッセージ変換：メッセージは EMQX Cloud のルール内で高度な処理や変換が可能で、Azure Blob Storage への書き込み前に適切な形式に整形できます。
  • 柔軟なデータ操作：Azure Blob Storage Sink により、特定のデータフィールドをコンテナに書き込むことが容易で、コンテナやオブジェクトキーの動的設定もサポートし柔軟なデータ保存が可能です。
  • 統合された業務プロセス：Azure Blob Storage Sink はデバイスデータを Azure Blob Storage の豊富なエコシステムと連携させ、データ分析やアーカイブなど多様なビジネスシナリオを実現します。
  • 低コストな長期保存：データベースと比較して、Azure Blob Storage は高可用性かつ信頼性の高いコスト効率の良いオブジェクトストレージサービスで、長期保存に適しています。

これらの特長により、効率的で信頼性が高くスケーラブルな IoT アプリケーションの構築と、ビジネスの意思決定や最適化に役立てることができます。

はじめる前に ​

本節では、EMQX で Azure Blob Storage Sink を作成する前の準備について説明します。

前提条件 ​

Azure ストレージの準備 ​

  1. Azure ストレージにアクセスするには Azure サブスクリプションが必要です。まだお持ちでない場合は、無料アカウントを作成してください。

  2. Azure ストレージへのアクセスはすべてストレージアカウントを通じて行われます。このクイックスタートでは、Azure ポータル、Azure PowerShell、または Azure CLI を使用してストレージアカウントを作成します。ストレージアカウントの作成方法はストレージアカウントの作成をご参照ください。

  3. Azure ポータルでコンテナを作成するには、新しく作成したストレージアカウントに移動します。ストレージアカウントの左メニューで「データストレージ」セクションまでスクロールし、「コンテナ」を選択します。+ コンテナ ボタンをクリックし、新しいコンテナ名に iot-data を入力して 作成 をクリックします。

    azure-storage-container-create

  4. ストレージアカウントの セキュリティ+ネットワーク -> アクセスキー に移動し、キー をコピーします。EMQX の Sink 設定時にこのキーが必要です。

    azure-storage-access-keys

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

デプロイ環境に適したネットワークアクセス方法を選択してください。

パブリックアクセス ​

  • 同一クラウドプラットフォームかつ同一リージョンの場合：NAT ゲートウェイを有効化せずにパブリックアクセスが可能です。
  • 異なるリージョンまたはクラウドプラットフォームの場合：パブリックアクセスにはNAT ゲートウェイを有効にしてください。

VNet ピアリングによるプライベートアクセス ​

コネクター作成前に以下のネットワーク設定を完了してください。

  1. EMQX Cloud のデプロイが Dedicated または Dedicated Flex エディションで、Azure 上で稼働し、Azure Blob Storage のプライベートエンドポイントを作成する VNet と同一リージョンであることを確認します。

  2. EMQX Cloud のデプロイと Azure VNet 間でVPC ピアリング接続を作成します。

  3. Azure ポータルでストレージアカウントを開き、ネットワーク -> プライベートエンドポイント接続に移動し、+ プライベートエンドポイントをクリックします。

  4. プライベートエンドポイントの作成ウィザードの基本ページで、サブスクリプションとリソースグループを選択し、他の必要なエンドポイント情報を入力して 次へ: リソース をクリックします。

  5. リソースページで以下を設定します：

    • 接続方法は ディレクトリ内の Azure リソースに接続 のままにします。
    • ストレージアカウントが含まれる サブスクリプション を選択します。
    • リソースの種類を Microsoft.Storage/storageAccounts に設定します。
    • リソースでストレージアカウントを選択します。
    • ターゲットサブリソースを blob に設定します。

    Select the Blob target sub-resource

    次へ: 仮想ネットワーク をクリックします。

  6. 仮想ネットワークページで、VPC ピアリング経由で EMQX Cloud デプロイと接続されている 仮想ネットワーク と サブネット を選択します。

    Select the virtual network and subnet

    次へ: DNS をクリックします。

  7. DNSページで、プライベート DNS 統合の プライベート DNS ゾーンと統合する を はい のままにし、デフォルトの privatelink.blob.core.windows.net プライベート DNS ゾーンを使用します。

    Configure private DNS integration

  8. 次へ: タグ をクリックし、必要に応じてタグを追加後、確認および作成 に進み、検証が成功したら 作成 をクリックします。

  9. プライベートエンドポイントのデプロイ完了後、プライベートエンドポイントリソースを開き、設定の DNS 構成を選択します。

  10. 以下の情報を控えます：

    • FQDN（例：<storage-account-name>.blob.core.windows.net）
    • プライベート IP アドレス

    View the private endpoint IP address and FQDN

  11. FQDN と IP アドレスを EMQX Cloud のサポートチケットで提出してください。EMQX Cloud の SRE チームが、EMQX Cloud デプロイがある VNet に対応するプライベート DNS レコードを追加します。

  12. 標準のストレージアカウントドメイン名がプライベートエンドポイントの IP アドレスに解決されることを確認してください。

  13. VNet ピアリング接続の両側で、ルートおよびネットワークセキュリティポリシーがポート 443 の HTTPS トラフィックを許可していることを確認してください。

コネクター設定時は、通常通りストレージアカウント名とアカウントキーを入力してください。プライベートエンドポイントの IP アドレスを入力する必要はありません。標準のストレージアカウントドメイン名を使用し、プライベート DNS によりプライベートエンドポイントの IP アドレスに解決されます。

プライベートアクセスのトラブルシューティング ​

VNet ピアリング経由でコネクターが Azure Blob Storage にアクセスできない場合、以下の一般的な問題を確認してください。

  • ストレージアカウントのドメイン名が解決できない。
  • ポート 443 の HTTPS トラフィックがプライベートエンドポイントに到達できない。
  • ストレージアカウントのネットワークポリシーが接続を拒否している。

コネクターの作成 ​

Azure Blob Storage Sink を追加する前に、対応するコネクターを作成する必要があります。

  1. デプロイメニューで データ統合 を選択し、データ永続化サービスカテゴリの中から Azure Blob Storage サービスを選択します。既に他のコネクターを作成している場合は、新規コネクター をクリックし、同じく Azure Blob Storage サービスを選択してください。

  2. コネクター名はシステムが自動的に生成します。

  3. 接続情報を入力します：

    • アカウント名：ストレージアカウント名
    • アカウントキー：前段で取得したストレージアカウントキー
    • 詳細設定（任意）：詳細設定を参照してください。

  4. 接続テストボタンをクリックし、Azure Blob Storage に正常にアクセスできれば成功メッセージが返されます。

    接続テストに失敗した場合、以下の一般的な問題を確認してください。

    • Azure Blob Storage が AuthorizationFailure エラーを返す。
    • ストレージアカウント名またはアカウントキーが誤っている。

  5. 作成ボタンをクリックしてコネクターの作成を完了します。

ルールの作成 ​

次に、書き込むデータを指定するルールを作成し、処理済みデータを Azure Blob Storage に転送する応答アクションを追加します。

  1. コネクター一覧の アクション 列にある新規ルールアイコンをクリックするか、ルール一覧 で 新規ルール をクリックし、新規ルール作成 ページに進みます。

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

    sql
    SELECT
  *
FROM
    "t/#"

    TIP

    SQL に不慣れな場合は、SQL 例 や 試してみる をクリックしてルール SQL の学習や結果のテストが可能です。

  3. 次へ をクリックしてアクション作成を開始します。

  4. コネクターを使用 のドロップダウンから、先ほど作成したコネクターを選択します。

  5. コンテナ に iot-data と入力します。

  6. アップロード方法 を選択します。2つの方法の違いは以下の通りです。

    • 直接アップロード：ルールがトリガーされるたびに、設定済みのオブジェクトキーとコンテンツに従ってデータを直接 Azure Storage にアップロードします。バイナリや大きなテキストデータの保存に適していますが、多数のファイルが生成される可能性があります。
    • 集約アップロード：複数回のルールトリガー結果を単一ファイル（例：CSV ファイル）にまとめて Azure Storage にアップロードします。構造化データの保存に適し、ファイル数を減らし書き込み効率を向上させます。

    各方法で設定項目が異なります。選択した方法に応じて設定してください。

  7. 必要に応じて詳細設定オプションを構成します（任意）。詳細は詳細設定をご参照ください。

  8. 確定ボタンをクリックしてアクション設定を完了します。

  9. 成功メッセージのポップアップで ルール一覧に戻る をクリックし、データ統合設定を完了します。

ルールのテスト ​

本節では、直接アップロード方式で設定したルールのテスト方法を示します。

  1. MQTTX を使ってトピック t/1 にメッセージをパブリッシュします。
bash
mqttx pub -i emqx_c -t t/1 -m '{ "msg": "Hello Azure" }'

  1. 数件のメッセージ送信後、Azure ポータルにログインし、ストレージアカウントの iot-data コンテナを開きます。アップロードされたオブジェクトがコンテナ内に表示されているはずです。

  2. EMQX Cloud コンソールで実行時データを確認します。ルール一覧でルール ID をクリックすると、ルールの統計情報やこのルールに紐づくすべてのアクションの実行時統計ページが表示されます。