MQTT データを Databricks に取り込む

ここで紹介するワークフローでは Amazon S3 を使用します。EMQX が MQTT メッセージを前処理し、S3 Sink を通じてファイルを書き込みます。Databricks は Unity Catalog の外部ロケーションを通じてそれらのファイルを読み込みます。

直接アップロードでは、設定した内容がそのまま保存されます。集約アップロードは、EMQX のバージョンに応じて CSV、JSON Lines、Parquet に対応しています。Databricks のワークフローで読み込める形式を選び、代表的なメッセージでテストしてください。

データの鮮度は、アップロードのバッチ設定、S3 への書き込み、Databricks がファイルを読み込み・処理するタイミングによって決まります。集約アップロードでは、レコード数と時間の上限を調整して、ファイルサイズとデータの鮮度のバランスを取ってください。

EMQX には、選択した S3 バケットとプレフィックスへの書き込み権限が必要です。Databricks には、IAM ベースのストレージ資格情報、外部ロケーション、およびそれらのファイルを読み込むために必要な Unity Catalog の権限が必要です。

はい。S3 Sink が Databricks 用のファイルを書き込む前に、EMQX の SQL ルールでトピックのマッチング、ペイロードフィールドの選択、メッセージのフィルタリングや変換を行えます。