DatabricksへのMQTTデータ取り込み
Databricksは、Apache Sparkを基盤とした統合データ分析プラットフォームであり、大規模なデータエンジニアリング、機械学習、協調分析を目的としています。EMQXは、Databricksが管理するAmazon S3バケットにMQTTデータを書き込み、Databricksが外部ロケーションを通じて直接クエリを実行できるようにすることで、Databricksと連携します。
本ページでは、EMQXとDatabricks間のデータ統合について詳しく解説し、コネクターおよびSinkの作成方法を実践的に案内します。
動作概要
EMQXのDatabricksデータ統合は、Amazon S3統合をベースに構築されています。EMQXはMQTTデータをDatabricksが管理するS3バケットに書き込み、Databricksは外部ロケーションを介してこのバケットにアクセスし、保存されたデータに対して直接SQLクエリを実行します。
具体的なワークフローは以下の通りです：
- デバイスのEMQXへの接続：IoTデバイスはMQTTプロトコルで正常に接続されるとオンラインイベントをトリガーします。このイベントにはデバイスID、送信元IPアドレス、その他のプロパティ情報が含まれます。
- デバイスからのメッセージパブリッシュと受信：デバイスは特定のトピックを通じてテレメトリやステータスデータをパブリッシュします。EMQXはメッセージを受信し、ルールエンジン内で比較処理を行います。
- ルールエンジンによるメッセージ処理：組み込みのルールエンジンは、トピックマッチングに基づいて特定のソースからのメッセージやイベントを処理します。対応するルールにマッチしたメッセージやイベントは、データフォーマット変換、特定情報のフィルタリング、コンテキスト情報によるメッセージの拡充などの処理が行われます。
- Amazon S3への書き込み：ルールはAmazon S3 Sinkをトリガーし、処理済みデータをDatabricksワークスペースに紐づくS3バケットに書き込みます。
- DatabricksによるS3からの読み込み：Databricksは外部ロケーションを介してS3バケットに保存されたデータを直接クエリし、リアルタイム分析や機械学習ワークフローを実現します。
特長とメリット
EMQXのDatabricksデータ統合を利用することで、以下のような特長と利点が得られます：
- メッセージ変換：メッセージはEMQXルール内で高度な処理や変換を経てからS3に書き込まれるため、後続の保存や分析が容易になります。
- 柔軟なデータ操作：Amazon S3 Sinkを用いることで、特定のデータフィールドをDatabricks管理のS3バケットに簡単に書き込め、動的なオブジェクトキー設定により柔軟なデータ保存が可能です。
- 統合分析プラットフォーム：EMQXとDatabricksの連携により、IoTデータが即座にDatabricksワークスペース内のSQL分析、機械学習、データエンジニアリングパイプラインで活用可能になります。
- 低コストの長期保存：S3を基盤ストレージとして利用することで、高可用性かつ信頼性の高いコスト効率の良いデータストアを実現し、大規模IoTワークロードに適しています。
はじめる前に
このセクションでは、EMQXでDatabricks用のAmazon S3コネクターおよびSinkを作成する前に必要な準備を紹介します。
前提条件
以下の内容に慣れていることを確認してください：
EMQXの概念：
Databricksの概念：
- ワークスペース：Databricksのすべての資産にアクセスする環境。
- 外部ロケーション：外部のS3パスをマッピングし、そこに保存されたデータをSQLで直接クエリ可能にするDatabricksの機能。
- ストレージ認証情報：外部ストレージロケーションの読み書き権限を付与するDatabricksのアクセス認証情報。
AWS MarketplaceでのDatabricksセットアップ
ここでは、AWS MarketplaceでのDatabricksサブスクリプションを例にデプロイ方法を説明します。
AWS MarketplaceでDatabricksをサブスクライブします。Databricksアカウントとワークスペース作成の案内が表示されます。
サブスクライブ後、ワークスペースを作成します。リージョンとストレージオプションを選択し、Createをクリックします。
ワークスペース作成後、Workspaces一覧に表示されます。ワークスペースに自動プロビジョニングされたS3バケット名（例：
databricks-workspace-stack-142ec-bucket）を控えてください。このバケットにEMQXのMQTTデータを保存します。
ワークスペースを開き、Catalog -> External locationsに移動し、EMQXがデータを書き込むS3パスを指す外部ロケーションを作成します。
Create locationをクリックし、Storage typeを
S3に設定、URLにs3://databricks-workspace-stack-142ec-bucket/emqx-iot-data-newを入力し、Storage credentialを選択します。
S3バケットの読み書き権限を持つIAMユーザーまたはロールのAWSアクセス認証情報（Access Key IDとSecret Access Key）を取得します。これらはEMQXコネクターの設定に使用します。
DatabricksワークスペースとS3バケットの設定が完了したら、EMQXでコネクターとSinkを作成する準備が整いました。
コネクターの作成
Amazon S3 Sinkを追加する前に、対応するコネクターを作成する必要があります。
- ダッシュボードのIntegration -> Connectorsページに移動します。
- 右上のCreateボタンをクリックします。
- コネクタータイプとしてAmazon S3を選択し、Nextをクリックします。
- コネクター名を入力します。名前は英数字で始まり、英数字、ハイフン、アンダースコアを含めることができます。この例では
my-databricksと入力します。 - 接続情報を入力します：
- Host：DatabricksワークスペースがデプロイされているAWSリージョンのS3エンドポイントを
s3.{region}.amazonaws.com形式で入力します。 - Port：
443を入力します。 - Access Key IDとSecret Access Key：AWS MarketplaceでのDatabricksセットアップで取得したAWSアクセス認証情報を入力します。
- Host：DatabricksワークスペースがデプロイされているAWSリージョンのS3エンドポイントを
- 残りの設定はデフォルト値を使用します。
- Createをクリックする前に、Test ConnectivityをクリックしてEMQXがS3サービスに接続できることを確認できます。
- Createをクリックしてコネクターの設定を完了します。作成成功のダイアログが表示され、ルールを今すぐ作成するか尋ねられます。Create Ruleをクリックするとコネクターが事前選択された状態でルール作成画面に進み、Back To Connector Listをクリックすると後でルールを作成できます。
Amazon S3 Sinkを使ったルールの作成
このセクションでは、EMQXでソースMQTTトピックt/#からのメッセージを処理し、処理結果をDatabricks管理のS3バケットに書き込むルールの作成方法を示します。
前のステップでCreate Ruleをクリックした場合、Add Actionパネルが自動で開き、Type of Actionが
Amazon S3、コネクターが事前選択されています。ステップ5へ進んでください。そうでなければ、ダッシュボードのIntegration -> Rulesページに移動し、右上のCreateをクリックします。ルールIDを入力し、SQLエディターに以下のルールSQLを入力します：
sqlSELECT * FROM "t/#"TIP
初心者の方は、SQL Examplesをクリックし、Enable Testを有効にしてSQLルールの学習とテストが可能です。
右側の**+ Add Actionをクリックし、Add ActionパネルでType of Action**ドロップダウンから
Amazon S3を選択し、ActionはデフォルトのCreate Actionのままにします。Connectorsドロップダウンから先ほど作成した
my-databricksコネクターを選択します。ドロップダウン横の作成ボタンをクリックするとポップアップで新規コネクターを素早く作成可能です。必要な設定パラメーターはコネクターの作成を参照してください。Sinkの名前と任意の説明を入力します。
Bucketに
databricks-workspace-stack-142ec-bucketを入力します。このフィールドは${var}形式のプレースホルダーもサポートしますが、対応するバケットがS3に存在することを確認してください。必要に応じてACLを選択し、アップロードされるオブジェクトのアクセス権限を指定します。
Upload Methodを選択します：
- Direct Upload：ルールがトリガーされるたびに、事前設定されたオブジェクトキーと内容に従ってデータを直接S3にアップロードします。バイナリや大きなテキストデータの保存に適しています。
- Aggregated Upload：複数のルールトリガー結果を1つのファイル（例：CSVファイル）にまとめてS3にアップロードします。構造化データの保存に適し、ファイル数を減らし書き込み効率を向上させます。
選択した方法により設定パラメーターが異なります。以下のタブで詳細を設定してください：
フォールバックアクション（任意）：メッセージ配信失敗時の信頼性向上のため、1つ以上のフォールバックアクションを定義できます。詳細はフォールバックアクションを参照してください。
Advanced Settingsを展開し、必要に応じて詳細設定を行います（任意）。詳細は高度な設定を参照してください。
残りの設定はデフォルト値を使用します。Createをクリックする前に、Test ConnectivityでSinkがS3サービスに接続できることを確認できます。
CreateをクリックしてSinkの作成を完了します。作成成功後、ルール作成画面に戻り、新しいSinkがルールアクションに追加されます。
ルール作成画面でSaveをクリックし、ルール作成プロセスを完了します。
これでルールの作成が完了しました。Rulesページで新規ルールを確認でき、**Actions (Sink)**タブで新しいAmazon S3 Sinkを確認できます。
ルールのテスト
MQTTXを使ってトピックt/1にメッセージをパブリッシュします：
mqttx pub -i emqx_c -t t/1 -m '{ "msg": "hello Databricks" }'数件メッセージを送信した後、DatabricksワークスペースでWorkspaceを右クリックし、Create -> Notebookを選択して新しいノートブックを作成します。
ノートブック内で外部ロケーションに対してSQLクエリを実行し、データが正常に取り込まれていることを確認します：
SELECT * FROM json.`s3://databricks-workspace-stack-142ec-bucket/emqx-iot-data-new/`
高度な設定
このセクションでは、Amazon S3 Sinkの高度な設定オプションについて説明します。ダッシュボードでSinkを設定する際、Advanced Settingsを展開して以下のパラメーターをニーズに応じて調整できます。
| フィールド名 | 説明 | デフォルト値 |
|---|---|---|
| Buffer Pool Size | EMQXとS3間のデータフローを管理するバッファワーカープロセスの数を指定します。 | 16 |
| Request TTL | リクエストがバッファに入ってから有効とみなされる最大時間（秒）を指定します。 | 45 |
| Health Check Interval | SinkがS3との接続状態を自動でヘルスチェックする間隔（秒）を指定します。 | 15秒 |
| Health Check Interval Jitter | 複数ノードが同時にヘルスチェックを開始する確率を減らすため、基本間隔に加える一様ランダム遅延（ミリ秒）です。 | 0ミリ秒 |
| Health Check Timeout | コネクターがS3との接続状態を自動でヘルスチェックする際のタイムアウト時間を指定します。 | 60秒 |
| Max Buffer Queue Size | S3 Sinkの各バッファワーカープロセスがバッファリング可能な最大バイト数を指定します。 | 256MB |
| Query Mode | メッセージ送信を最適化するために、synchronous（同期）またはasynchronous（非同期）リクエストモードを選択できます。 | Asynchronous |
| In-flight Window | SinkがS3と通信中に同時に存在可能なインフライトキューリクエストの最大数を制御します。 | 100 |
| Min Part Size | 集約完了後のパートアップロードの最小チャンクサイズを指定します。 | 5MB |
| Max Part Size | パートアップロードの最大チャンクサイズを指定します。 | 5GB |