DynamoDBへのMQTTデータ取り込み

DynamoDBは、AWS上で提供されるフルマネージドの高性能サーバーレスなキー・バリュー型ストアデータベースサービスです。高速でスケーラブルかつ信頼性の高いデータストレージを必要とするアプリケーションに適しています。EMQXはDynamoDBとの連携をサポートしており、MQTTメッセージやクライアントイベントをDynamoDBに保存することで、IoTデバイスの登録・管理やデバイスデータの長期保存およびリアルタイム分析を実現します。DynamoDBのデータ統合を通じて、MQTTメッセージやクライアントイベントをDynamoDBに格納できるほか、イベントに応じてDynamoDB内のデータの更新や削除をトリガーし、デバイスのオンライン状態や接続履歴などの情報を記録できます。

本ページでは、EMQXとDynamoDB間のデータ統合について包括的に解説し、データ統合の作成および検証方法を実践的に説明します。

動作概要

DynamoDBデータ統合は、EMQXに標準搭載された機能であり、EMQXのデバイス接続およびメッセージ送受信機能とDynamoDBの強力なデータストレージ機能を組み合わせています。組み込みのルールエンジンコンポーネントにより、EMQXからDynamoDBへのデータ取り込みを簡素化し、複雑なコーディングを不要にします。

以下の図は、EMQXとDynamoDB間のデータ統合の典型的なアーキテクチャを示しています。

EMQX Integration DynamoDB

MQTTデータのDynamoDBへの取り込みは、以下のように動作します。

メッセージのパブリッシュと受信：接続された車両、IIoTシステム、エネルギー管理プラットフォームなどのIoTデバイスは、MQTTプロトコルを介してEMQXに正常に接続し、特定のトピックにMQTTメッセージをパブリッシュします。EMQXがこれらのメッセージを受信すると、ルールエンジン内でマッチング処理を開始します。
メッセージデータの処理：メッセージが到着すると、ルールエンジンを通過し、EMQXで定義されたルールに従って処理されます。ルールは事前定義された条件に基づき、DynamoDBにルーティングすべきメッセージを判定します。ペイロード変換が指定されている場合は、データ形式の変換、特定情報のフィルタリング、追加コンテキストによるペイロードの拡充などが適用されます。
DynamoDBへのデータ取り込み：ルールエンジンがDynamoDBへの保存対象メッセージを特定すると、DynamoDBへの転送アクションをトリガーします。処理済みデータはDynamoDBのテーブルにシームレスに書き込まれます。
データの保存と活用：データがDynamoDBに保存された後、企業はそのクエリ機能を活用して様々なユースケースに利用できます。例えば、接続車両分野では、車両の状態管理、リアルタイム指標に基づくルート最適化、資産追跡などに活用可能です。IIoT環境では、機械の状態監視、メンテナンス予測、生産スケジュールの最適化などに利用されます。

特長とメリット

DynamoDBとのデータ統合は、効率的なデータ送信、保存、活用を実現するための多彩な特長とメリットを提供します。

リアルタイムデータストリーミング：EMQXはリアルタイムデータストリームの処理に最適化されており、ソースシステムからDynamoDBへの効率的かつ信頼性の高いデータ送信を保証します。即時のインサイトやアクションが必要なユースケースに適しています。
柔軟なデータ変換：EMQXは強力なSQLベースのルールエンジンを提供し、DynamoDBに保存する前にデータを前処理できます。フィルタリング、ルーティング、集約、拡充など多様な変換機構をサポートし、ニーズに応じたデータ整形が可能です。
柔軟なデータモデル：DynamoDBはキー・バリュー型およびドキュメント型データモデルを採用しており、構造化されたデバイスイベントやメッセージデータの保存・管理に適しています。異なるMQTTメッセージ構造も容易に格納できます。
強力なスケーラビリティ：EMQXはクラスターのスケーラビリティを提供し、デバイス接続数やメッセージ量に応じてシームレスな水平スケールが可能です。DynamoDBはサーバーやインフラ管理を不要とし、リソース管理とスケーリングを自動で行います。両者の組み合わせにより、高性能かつ高信頼性のデータ保存とスケーラビリティを実現します。

はじめる前に

このセクションでは、DynamoDBデータ統合を作成する前に必要な準備について説明します。認証方法の選択、DynamoDBサーバーのインストール、データテーブルの作成が含まれます。

前提条件

EMQXのデータ統合に関するルールの知識
データ統合の知識

認証方法の選択

EMQX 6.0.4以降、DynamoDBコネクターは以下の認証方法をサポートしています。EMQXのデプロイ環境に応じて適切な方法を選択してください。

アクセスキーの手動設定：対象のDynamoDBリソースにアクセス権限を持つAWS Access Key IDとAWS Secret Access Keyを指定します。この方法はローカルデプロイ、AWS外環境、ECSタスクロールやEC2インスタンスロールを使用しないデプロイに適しています。
一時認証情報の自動取得：EMQXがAmazon ECSタスクまたはAmazon EC2インスタンスとして実行されている場合、対象のDynamoDBリソースにアクセス権限を持つECSタスクロールまたはEC2インスタンスロールを設定します。コネクターのAWS Access Key IDとAWS Secret Access Keyは空欄にします。EMQXはECSタスクロールまたはEC2インスタンスメタデータから一時認証情報を取得し、有効期限前に更新します。

重要なお知らせ

AWS Access Key IDとAWS Secret Access Keyは両方とも指定するか、両方とも空欄にする必要があります。一方のみを指定すると、コネクター設定が無効になります。

DynamoDBローカルサーバーのインストールとテーブル作成

以下のコマンドでDynamoDBサーバーをローカルで起動します。

Access Key ID: root
Secret Access Key: public
Region: us-west-2

bash

docker run -d -p 8000:8000 --name dynamodb-local \
  -e AWS_ACCESS_KEY_ID=root \
  -e AWS_SECRET_ACCESS_KEY=public \
  -e AWS_DEFAULT_REGION=us-west-2 \
  amazon/dynamodb-local:2.4.0

テーブル定義ファイルを作成し、カレントディレクトリにmqtt_msg.jsonという名前で保存します。テーブル定義は以下の通りです。

device_idをハッシュキー（パーティションキー）として定義。
timestampをレンジキー（ソートキー）として定義。
device_id属性を文字列型（S）として定義。
timestamp属性を数値型（N）として定義。

json

{
    "TableName": "mqtt_msg",
    "AttributeDefinitions": [
        {
            "AttributeName": "device_id",
            "AttributeType": "S"
        },
        {
            "AttributeName": "timestamp",
            "AttributeType": "N"
        }
    ],
    "KeySchema": [
        {
            "AttributeName": "device_id",
            "KeyType": "HASH"
        },
        {
            "AttributeName": "timestamp",
            "KeyType": "RANGE"
        }
    ],
    "ProvisionedThroughput": {
        "ReadCapacityUnits": 5,
        "WriteCapacityUnits": 5
    }
}

Dockerを使ってaws-cliコマンドを実行し、ファイルを用いて新しいテーブルを作成します。

bash

docker run --rm -v $PWD:/dynamo_data \
    -e AWS_ACCESS_KEY_ID=root \
    -e AWS_SECRET_ACCESS_KEY=public \
    -e AWS_DEFAULT_REGION=us-west-2 \
    amazon/aws-cli:2.15.57 dynamodb create-table \
    --cli-input-json file:///dynamo_data/mqtt_msg.json \
    --endpoint-url http://host.docker.internal:8000

Dockerを使ってaws-cliコマンドを実行し、テーブル作成が成功したか確認します。

bash

docker run --rm \
    -e AWS_ACCESS_KEY_ID=root \
    -e AWS_SECRET_ACCESS_KEY=public \
    -e AWS_DEFAULT_REGION=us-west-2 \
    amazon/aws-cli:2.15.57 dynamodb list-tables \
    --endpoint-url http://host.docker.internal:8000

テーブル作成が成功していれば、以下のJSONが表示されます。

json

{
    "TableNames": [
        "mqtt_msg"
    ]
}

コネクターの作成

このセクションでは、SinkをDynamoDBサーバーに接続するコネクターの作成方法を説明します。

以下の手順は、EMQXとDynamoDBをローカルマシン上で実行していることを前提としています。リモートで実行している場合は設定を適宜調整してください。

EMQXダッシュボードにアクセスし、Integration -> Connectorsをクリックします。
ページ右上のCreateをクリックします。
Create ConnectorページでDynamoDBを選択し、Nextをクリックします。
Configurationステップで以下の情報を設定します。
- Connector name：コネクター名を入力します。大文字・小文字の英数字の組み合わせが推奨されます。例：my_dynamodb
- DynamoDB Region：us-west-2を入力します。
- DynamoDB Endpoint：http://127.0.0.1:8000（ローカルの場合）またはDynamoDBサーバーの実際のURLを入力します。
- AWS Access Key IDおよびAWS Secret Access Key：ローカルDynamoDBの例ではそれぞれrootとpublicを入力します。ECSタスクロールやEC2インスタンスロールを使用する場合は両方空欄にします。詳細は認証方法の選択を参照してください。
詳細設定（任意）：詳細はSinkの特長を参照してください。
Createをクリックする前に、Test ConnectivityをクリックしてコネクターがDynamoDBサーバーに接続可能かテストできます。
ページ下部のCreateボタンをクリックしてコネクターの作成を完了します。ポップアップダイアログでBack to Connector Listをクリックするか、Create RuleをクリックしてSinkを使ったルール作成を続行できます。詳細はメッセージ保存用DynamoDB Sinkルールの作成およびイベント記録用DynamoDB Sinkルールの作成を参照してください。

メッセージ保存用DynamoDB Sinkルールの作成

このセクションでは、ソースMQTTトピックt/#からのメッセージを処理し、処理済みデータを設定済みSink経由でDynamoDBテーブルmqtt_msgに書き込むルールの作成方法をダッシュボード上で説明します。

EMQXダッシュボードにアクセスし、Integration -> Rulesをクリックします。
ページ右上のCreateをクリックします。
ルールIDにmy_ruleを入力します。メッセージ保存用ルールを作成するため、SQL Editorに以下のステートメントを入力します。これはトピックt/#配下のMQTTメッセージをDynamoDBに保存することを意味します。
注意：独自のSQL構文を指定する場合は、Sinkが必要とするすべてのフィールドをSELECT部に含めていることを確認してください。
sql
```
SELECT 
  *
FROM
  "t/#"
```
TIP
初心者の方は、SQL Examplesをクリックし、Enable Testを有効にしてSQLルールを学習・テストできます。
- Add Actionボタンをクリックして、ルールによってトリガーされるアクションを定義します。このアクションにより、EMQXはルールで処理したデータをDynamoDBに送信します。
Type of ActionのドロップダウンリストからDynamoDBを選択します。ActionはデフォルトのCreate Actionのままにします。既に作成済みのSinkを選択することもできますが、ここでは新規Sinkを作成します。
Sinkの名前を入力します。名前は大文字・小文字の英数字の組み合わせにしてください。
Connectorのドロップダウンから先ほど作成したmy_dynamodbを選択します。隣のボタンから新規コネクター作成も可能です。設定パラメータの詳細はコネクターの作成を参照してください。
以下の設定を行います。
- Table：先に作成したテーブル名mqtt_msgを入力します。
- Hash Key：${clientid}を入力し、クライアントIDをハッシュキーとして使用します。
- Range Key（任意）：${timestamp}を入力し、メッセージのタイムスタンプをレンジキーとして使用します。
- Message Template：デフォルトでは空欄のままにします。
  TIP
  この値が空欄の場合、メッセージ全体がデータベースに保存されます。実際の値はJSONテンプレートデータです。
  SQLテンプレート内でプレースホルダー変数が未定義の場合、Message template上部のUndefined Vars as Nullスイッチでルールエンジンの動作を切り替えられます。
  - Disabled（デフォルト）：ルールエンジンは文字列undefinedをデータベースに挿入します。
  - Enabled：未定義変数の場合、ルールエンジンはNULLをデータベースに挿入します。
    TIP
    可能な限りこのオプションは有効にしてください。無効化は後方互換性確保のためのみ推奨されます。
フォールバックアクション（任意）：メッセージ配送失敗時の信頼性向上のため、1つ以上のフォールバックアクションを定義できます。これらはプライマリSinkがメッセージ処理に失敗した場合にトリガーされます。詳細はフォールバックアクションを参照してください。
詳細設定（任意）：必要に応じてsyncまたはasyncクエリモードを選択します。詳細はSinkの特長を参照してください。
Createをクリックする前に、Test ConnectivityをクリックしてSinkがサーバーに接続可能かテストできます。
CreateボタンをクリックしてSink設定を完了します。新しいSinkがAction Outputsに追加されます。
Create Ruleページに戻り、設定内容を確認してからCreateボタンをクリックし、ルールを生成します。

これでDynamoDB Sinkを通じてデータを転送するルールの作成が完了しました。Integration -> Rulesページで新規作成したルールを確認できます。**Actions(Sink)**タブをクリックすると、新しいDynamoDB Sinkが表示されます。

また、Integration -> Flow Designerをクリックするとトポロジーを確認でき、トピックt/#配下のメッセージがルールmy_ruleで解析されDynamoDBに送信・保存されている様子が確認できます。

イベント記録用DynamoDB Sinkルールの作成

このセクションでは、クライアントのオンライン／オフライン状態を記録し、イベントデータを設定済みSink経由でDynamoDBテーブルmqtt_msgに書き込むルールの作成方法を説明します。

TIP

便宜上、オンライン／オフラインイベントの受信にはmqtt_msgトピックを再利用します。

ルールおよびアクションの作成手順は、メッセージ保存用DynamoDB Sinkルールの作成とほぼ同様ですが、SQLルール構文が異なります。

オンライン／オフライン状態記録用のSQLルール構文は以下の通りです。

sql

SELECT
  str(event) + timestamp as id, *
FROM 
  "$events/client_connected", "$events/client_disconnected"

ルールのテスト

MQTT Xを使ってトピックt/1にメッセージを送信し、オンライン／オフラインイベントをトリガーします。

bash

mqttx pub -i emqx_c -t t/1 -m '{ "msg": "hello DynamoDB" }'

Sinkの稼働状況を確認すると、1件の新規受信メッセージと1件の新規送信メッセージ、合計2件のイベントレコードがあるはずです。

mqtt_msgデータテーブルにデータが書き込まれているか確認します。

bash

docker run --rm -e AWS_ACCESS_KEY_ID=root -e AWS_SECRET_ACCESS_KEY=public -e AWS_DEFAULT_REGION=us-west-2 amazon/aws-cli dynamodb scan --table-name=mqtt_msg --endpoint-url http://host.docker.internal:8000

出力例：

json

{
    "Items": [
        {
            "metadata": {
                "S": "{\"rule_id\":\"90d98f59\"}"
            },
            "peerhost": {
                "S": "127.0.0.1"
            },
            "clientid": {
                "S": "emqx_c"
            },
            "flags": {
                "S": "{\"retain\":false,\"dup\":false}"
            },
            "node": {
                "S": "emqx@127.0.0.1"
            },
            "qos": {
                "N": "0"
            },
            "payload": {
                "S": "{ \"msg\": \"hello DynamoDB\" }"
            },
            "pub_props": {
                "S": "{\"User-Property\":{}}"
            },
            "publish_received_at": {
                "N": "1678263363503"
            },
            "topic": {
                "S": "t/1"
            },
            "id": {
                "S": "0005F65F239F03FEF44300000BB40002"
            },
            "event": {
                "S": "message.publish"
            },
            "username": {
                "S": "undefined"
            },
            "timestamp": {
                "N": "1678263363503"
            }
        },
        {
            "conn_props": {
                "S": "{\"User-Property\":{},\"Request-Problem-Information\":1}"
            },
            "peername": {
                "S": "127.0.0.1:59582"
            },
            "metadata": {
                "S": "{\"rule_id\":\"703890a5\"}"
            },
            "clientid": {
                "S": "emqx_c"
            },
            "is_bridge": {
                "S": "false"
            },
            "keepalive": {
                "N": "30"
            },
            "proto_ver": {
                "N": "5"
            },
            "proto_name": {
                "S": "MQTT"
            },
            "connected_at": {
                "N": "1678263363499"
            },
            "receive_maximum": {
                "N": "32"
            },
            "sockname": {
                "S": "127.0.0.1:1883"
            },
            "mountpoint": {
                "S": "undefined"
            },
            "node": {
                "S": "emqx@127.0.0.1"
            },
            "id": {
                "S": "client.connected1678263363499"
            },
            "expiry_interval": {
                "N": "0"
            },
            "event": {
                "S": "client.connected"
            },
            "username": {
                "S": "undefined"
            },
            "timestamp": {
                "N": "1678263363499"
            },
            "clean_start": {
                "S": "true"
            }
        },
        {
            "reason": {
                "S": "normal"
            },
            "peername": {
                "S": "127.0.0.1:59582"
            },
            "metadata": {
                "S": "{\"rule_id\":\"703890a5\"}"
            },
            "clientid": {
                "S": "emqx_c"
            },
            "proto_ver": {
                "N": "5"
            },
            "proto_name": {
                "S": "MQTT"
            },
            "sockname": {
                "S": "127.0.0.1:1883"
            },
            "disconn_props": {
                "S": "{\"User-Property\":{}}"
            },
            "node": {
                "S": "emqx@127.0.0.1"
            },
            "id": {
                "S": "client.disconnected1678263363503"
            },
            "event": {
                "S": "client.disconnected"
            },
            "disconnected_at": {
                "N": "1678263363503"
            },
            "username": {
                "S": "undefined"
            },
            "timestamp": {
                "N": "1678263363503"
            }
        }
    ],
    "Count": 3,
    "ScannedCount": 3,
    "ConsumedCapacity": null
}

Kubernetes

EMQX Operator

EMQX クラスター管理

API リファレンス

パスワード認証

監視

アクセスコントロール

統合

管理

Prometheus との統合

OpenTelemetry との統合

エンドツーエンドトレース

プラグインカタログ

JT/T 808 ゲートウェイ

GB/T 32960 ゲートウェイ

Volcengine RTC

シナリオ

GPT-Realtime

DynamoDBへのMQTTデータ取り込み ​

動作概要 ​

特長とメリット ​

はじめる前に ​

前提条件 ​

認証方法の選択 ​

DynamoDBローカルサーバーのインストールとテーブル作成 ​

コネクターの作成 ​

メッセージ保存用DynamoDB Sinkルールの作成 ​

イベント記録用DynamoDB Sinkルールの作成 ​

ルールのテスト ​

DynamoDBへのMQTTデータ取り込み

動作概要

特長とメリット

はじめる前に

前提条件

認証方法の選択

DynamoDBローカルサーバーのインストールとテーブル作成

コネクターの作成

メッセージ保存用DynamoDB Sinkルールの作成

イベント記録用DynamoDB Sinkルールの作成

ルールのテスト