Amazon KinesisへのMQTTデータストリーム

AWS Kinesisは、AWS上のフルマネージドリアルタイムストリーミングデータ処理サービスであり、ストリーミングデータの収集、処理、分析を容易に行えます。あらゆる規模のストリーミングデータを経済的かつ効率的にリアルタイムで処理でき、高い柔軟性を持ち、数十万のソースからの大量のストリーミングデータを低レイテンシで処理可能です。

EMQXはAmazon Kinesis Data Streamsとのシームレスな統合をサポートしており、大規模なIoTデバイスのリアルタイムメッセージ収集と送信を実現します。このデータ統合により、Amazon Kinesis Data Streamsと接続してリアルタイムデータ分析や複雑なストリーム処理が可能です。

本ページでは、EMQXとAmazon Kinesis間のデータ統合について包括的に紹介し、実際の作成および検証手順を解説します。

動作原理

Amazon Kinesisデータ統合は、EMQXの標準機能として提供されており、ユーザーがMQTTデータストリームをAmazon Kinesisにシームレスに統合し、その豊富なサービスと機能を活用してIoTアプリケーションを開発できるよう支援します。

emqx-integration-aws

EMQXはルールエンジンとSinkを介してMQTTデータをAmazon Kinesisに転送します。全体の流れは以下の通りです。

IoTデバイスがメッセージをパブリッシュ：デバイスは特定のトピックを通じてテレメトリや状態データをパブリッシュし、ルールエンジンをトリガーします。
ルールエンジンがメッセージを処理：組み込みのルールエンジンは特定のトピックにマッチするMQTTメッセージを処理します。ルールにマッチしたメッセージは、データ形式の変換、特定情報のフィルタリング、コンテキスト情報の付加などの処理が行われます。
Amazon Kinesisへのブリッジング：ルールによってトリガーされたアクションでメッセージをAmazon Kinesisに転送します。パーティションキー、書き込み先のデータストリーム、メッセージフォーマットをカスタマイズ可能で、柔軟なデータ統合を実現します。

MQTTメッセージデータがAmazon Kinesisに書き込まれた後は、以下のような柔軟なアプリケーション開発が可能です。

リアルタイムデータ処理・分析：Amazon Kinesisの強力なデータ処理・分析ツールとストリーミング機能を活用し、メッセージデータのリアルタイム処理・分析を行い、有益なインサイトや意思決定支援を得られます。
イベント駆動型機能：Amazonのイベント処理をトリガーし、動的かつ柔軟な機能トリガーと処理を実現します。
データ保存・共有：メッセージデータをAmazon Kinesisのストレージサービスに送信し、大量データの安全な保存・管理を行います。これにより他のAmazonサービスと連携してデータの共有・分析が可能となり、多様なビジネスニーズに対応します。

特長とメリット

EMQXとAWS Kinesis Data Streamsのデータ統合は、以下の機能と利点をビジネスにもたらします。

信頼性の高いデータ送信と順序保証：EMQXとAWS Kinesis Data Streamsは共に信頼性の高いデータ送信機構を備えています。EMQXはMQTTプロトコルを通じてメッセージの確実な送信を保証し、AWS Kinesis Data Streamsはパーティションとシーケンス番号によりメッセージの順序を保証します。これにより、デバイスから送信されたメッセージが正確に目的地に届き、正しい順序で処理されます。
リアルタイムデータ処理：デバイスからの高頻度データはEMQXのルールSQLで事前にリアルタイム処理が可能で、MQTTメッセージのフィルタリング、抽出、付加、変換を容易に行えます。AWS Kinesis Data Streamsに送信後は、AWS LambdaやAWS管理のApache Flinkと組み合わせてさらなるリアルタイム分析が実現できます。
弾力的なスケーラビリティ対応：EMQXは数百万のIoTデバイス接続を容易に実現し、弾力的なスケーラビリティを提供します。一方、AWS Kinesis Data Streamsはオンデマンドの自動リソース割り当てと拡張を行います。両者を組み合わせたアプリケーションは接続数やデータ量の増加に応じてスケールし、ビジネスの成長に継続的に対応します。
パーシステンス（永続化）されたデータ保存：AWS Kinesis Data Streamsはパーシステンス機能を備え、毎秒数百万のデバイスデータストリームを確実に保存します。必要に応じて過去データの取得が可能で、オフライン分析や処理にも対応します。

AWS Kinesis Data Streamsを利用したストリーミングデータパイプラインの構築により、EMQXとAWSプラットフォームの統合の難易度が大幅に低減され、ユーザーにより豊富で柔軟なデータ処理ソリューションを提供します。これにより、EMQXユーザーはAWS上で機能的に充実した高性能なデータ駆動型アプリケーションを構築できます。

はじめる前に

本節では、Amazon Kinesisデータ統合の作成を始める前に必要な準備事項を説明します。Kinesisサービスのセットアップやデータストリームサービスのエミュレーション方法も含みます。

前提条件

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

Amazon Kinesis Data Streamsでのストリーム作成

以下の手順に従い、AWSマネジメントコンソールからストリームを作成します（詳細はこちらのチュートリアルを参照）。

AWSマネジメントコンソールにサインインし、Kinesisコンソールを開きます。
ナビゲーションバーのリージョンセレクターを展開し、リージョンを選択します。
Create data streamを選択します。
Create Kinesis streamページでデータストリーム名を入力し、On-demandキャパシティモードを選択します。

Amazon Kinesis Data Streamsのローカルエミュレーション

開発やテストを容易にするため、LocalStackを使ってAmazon Kinesis Data Streamsサービスをローカルでエミュレートできます。LocalStackを使うと、リモートクラウドに接続せずにローカルマシン上でAWSアプリケーションを完全に実行可能です。

Dockerイメージを使ってインストール・起動します。

bash

# LocalStackのDockerイメージをローカルで起動
docker run --name localstack -p '4566:4566' -e 'KINESIS_LATENCY=0' -d localstack/localstack:2.1

# コンテナにアクセス
docker exec -it localstack bash

シャード数1でmy_streamという名前のストリームを作成します。
bash
```
awslocal kinesis create-stream --stream-name "my_stream" --shard-count 1
```

コネクターの作成

本節では、SinkをAmazon Kinesis Data Streamsサービスに接続するコネクターの作成方法を説明します。

EMQXダッシュボードに入り、Integration -> Connectorsをクリックします。
ページ右上のCreateをクリックします。
Create ConnectorページでAmazon Kinesisを選択し、Nextをクリックします。
Configurationステップで以下を設定します。
- コネクター名を入力します。英数字の組み合わせで、例：my_kinesis。
- Amazon Kinesis Endpoint：Kinesisサービスのエンドポイントを入力します。LocalStackを使用する場合はhttp://localhost:4566を入力してください。
- AWS Access Key ID：アクセスキーIDを入力します。LocalStack使用時は任意の値で構いません。
- AWS Secret Access Key：シークレットアクセスキーを入力します。LocalStack使用時は任意の値で構いません。
Createをクリックする前に、Test Connectivityを押してコネクターがAmazon Kinesis Data Streamsに接続可能かテストできます。
ページ下部のCreateボタンをクリックしてコネクター作成を完了します。ポップアップダイアログでBack to Connector Listをクリックするか、Create RuleをクリックしてルールとSinkの作成を続行できます。詳細はAmazon Kinesis Sink付きルールの作成を参照してください。

Amazon Kinesis Sink付きルールの作成

本節では、ソースMQTTトピックt/#からのメッセージを処理し、処理結果をAmazonデータストリームmy_streamにストリーミングするルールの作成方法を示します。

EMQXダッシュボードでIntegration -> Rulesをクリックします。
ページ右上のCreateをクリックします。
ルールIDにmy_ruleを入力します。
SQL Editorでルールを設定します。トピックt/#のMQTTメッセージをAmazon Kinesis Data Streamsに保存したい場合、以下のSQL文を使用できます。
注意：独自のSQL文を指定する場合は、Sinkのペイロードテンプレートで必要なすべてのフィールドをSELECTに含めてください。
sql
```
SELECT
  *
FROM
  "t/#"
```
TIP
初心者の方はSQL Examplesをクリックし、Enable TestでSQLルールの学習とテストが可能です。
- Add Actionボタンをクリックし、ルールによってトリガーされるアクションを定義します。このアクションにより、EMQXはルールで処理したデータをKinesisに送信します。
Type of ActionドロップダウンからAmazon Kinesisを選択します。ActionドロップダウンはデフォルトのCreate Actionのままにします。既に作成済みのSinkを選択することも可能ですが、本例では新規Sinkを作成します。
Sinkの名前と説明を入力します。名前は英数字の組み合わせにしてください。
Connectorドロップダウンから先に作成したmy_kinesisを選択します。新規コネクターを作成する場合はドロップダウン横のボタンをクリックしてください。設定パラメータはコネクターの作成を参照。
以下の情報を入力します。
- Amazon Kinesis Stream：Amazon Kinesis Data Streamsでのストリーム作成で作成したストリーム名を入力。
- Partition Key：このストリームに送信されるレコードに関連付けるパーティションキーを入力。${variable_name}形式のプレースホルダーも使用可能（次ステップの例参照）。
Payload Template欄は空白のままにするか、テンプレートを定義します。
- 空白の場合、MQTTメッセージのクライアントID、トピック、ペイロードなどの可視入力をすべてJSON形式でエンコードします。
- 定義したテンプレートを使う場合、${variable_name}形式のプレースホルダーはMQTTコンテキストの対応値に置換されます。例：${topic}はMQTTメッセージのトピックがmy/topicならmy/topicに置換されます。
フォールバックアクション（任意）：メッセージ配信失敗時の信頼性向上のため、1つ以上のフォールバックアクションを定義可能です。これらはプライマリSinkがメッセージ処理に失敗した場合にトリガーされます。詳細はフォールバックアクションを参照してください。
詳細設定（任意）：必要に応じて詳細設定オプションを構成します。詳細は詳細設定を参照してください。
Createをクリックする前に、Test Connectivityを押してSinkがAmazon Kinesis Data Streamsに接続可能かテストできます。
CreateボタンをクリックしてSink設定を完了します。新しいSinkがAction Outputsに追加されます。
Create Ruleページに戻り、設定内容を確認してCreateをクリックしルールを生成します。

これでAmazon Kinesis Sinkを介してデータを転送するルールが正常に作成されました。Integration -> Rulesページで新規作成ルールを確認できます。**Actions(Sink)**タブをクリックすると新しいAmazon Kinesis Sinkが表示されます。

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

ルールのテスト

MQTTXを使い、トピックt/my_topicにメッセージを送信します。
bash
```
mqttx pub -i emqx_c -t t/my_topic -m '{ "msg": "hello Amazon Kinesis" }'
```
Sinkの稼働状況を確認すると、新規の受信メッセージと送信メッセージが1件ずつあるはずです。
Amazon Kinesis Data Viewerにアクセスし、レコード取得時にメッセージが表示されることを確認します。

LocalStackでの確認方法

LocalStackを利用している場合は、以下の手順で受信データを確認します。

メッセージ送信前に、以下のコマンドでShardIteratorを取得します。

bash

awslocal kinesis get-shard-iterator --stream-name my_stream --shard-id shardId-000000000000 --shard-iterator-type LATEST
{
"ShardIterator": "AAAAAAAAAAG3YjBK9sp0uSIFGTPIYBI17bJ1RsqX4uJmRllBAZmFRnjq1kPLrgcyn7RVigmH+WsGciWpImxjXYLJhmqI2QO/DrlLfp6d1IyJFixg1s+MhtKoM6IOH0Tb2CPW9NwPYoT809x03n1zL8HbkXg7hpZjWXPmsEvkXjn4UCBf5dBerq7NLKS3RtAmOiXVN6skPpk="
}

MQTTXを使い、トピックt/my_topicにメッセージを送信します。
bash
```
mqttx pub -i emqx_c -t t/my_topic -m '{ "msg": "hello Amazon Kinesis" }'
```

レコードを読み取り、受信データをデコードします。

bash

awslocal kinesis get-records --shard-iterator="AAAAAAAAAAG3YjBK9sp0uSIFGTPIYBI17bJ1RsqX4uJmRllBAZmFRnjq1kPLrgcyn7RVigmH+WsGciWpImxjXYLJhmqI2QO/DrlLfp6d1IyJFixg1s+MhtKoM6IOH0Tb2CPW9NwPYoT809x03n1zL8HbkXg7hpZjWXPmsEvkXjn4UCBf5dBerq7NLKS3RtAmOiXVN6skPpk="
{
    "Records": [
        {
            "SequenceNumber": "49642650476690467334495639799144299020426020544120356866",
            "ApproximateArrivalTimestamp": 1689389148.261,
            "Data": "eyAibXNnIjogImhlbGxvIEFtYXpvbiBLaW5lc2lzIiB9",
            "PartitionKey": "key",
            "EncryptionType": "NONE"
        }
    ],
    "NextShardIterator": "AAAAAAAAAAFj5M3+6XUECflJAlkoSNHV/LBciTYY9If2z1iP+egC/PtdVI2t1HCf3L0S6efAxb01UtvI+3ZSh6BO02+L0BxP5ssB6ONBPfFgqvUIjbfu0GOmzUaPiHTqS8nNjoBtqk0fkYFDOiATdCCnMSqZDVqvARng5oiObgigmxq8InciH+xry2vce1dF9+RRFkKLBc0=",
    "MillisBehindLatest": 0
}

echo 'eyAibXNnIjogImhlbGxvIEFtYXpvbiBLaW5lc2lzIiB9' | base64 -d
{ "msg": "hello Amazon Kinesis" }

詳細設定

本節では、Amazon Kinesis Sinkの詳細設定オプションについて説明します。ダッシュボードのSink設定画面でAdvanced Settingsを展開し、用途に応じて以下のパラメータを調整できます。

フィールド名	説明	デフォルト値
Buffer Pool Size	EMQXとKinesis間のデータフローを管理するバッファワーカーの数を指定します。これらのワーカーはデータを一時的に保持・処理し、ターゲットサービスへの送信を最適化しスムーズなデータ転送を保証します。	`16`
Request TTL	バッファに入ったリクエストが有効とみなされる最大時間（秒）を指定します。リクエストがこのTTLを超えてバッファ内にある、または送信後にKinesisからの応答やアックを受け取れなかった場合、リクエストは期限切れと判断されます。	`45`秒
Health Check Interval	SinkがKinesisとの接続状態を自動的にヘルスチェックする間隔（秒）を指定します。	`15`秒
Health Check Interval Jitter	複数ノードが同時にヘルスチェックを開始する確率を減らすため、基本のヘルスチェック間隔に加える一様ランダム遅延です。複数のアクションやソースが同じコネクターを共有する場合、ジッターを有効にするとヘルスチェックが少しずつずれて実行されます。	`15`秒
Health Check Timeout	コネクターがKinesisとの接続ヘルスチェックを行う際のタイムアウト時間（秒）を指定します。	`60`秒
Max Buffer Queue Size	Kinesis Sinkの各バッファワーカーがバッファリング可能な最大バイト数を指定します。バッファワーカーはデータを一時的に保持し、Kinesisへの送信を効率化します。システム性能やデータ転送要件に応じて調整してください。	`256`
Query Mode	`synchronous`または`asynchronous`のリクエストモードを選択し、メッセージ送信を最適化します。非同期モードではKinesisへの書き込みがMQTTメッセージのパブリッシュをブロックしませんが、クライアントがKinesis到達前にメッセージを受信する可能性があります。	`Async`
Batch Size	EMQXからKinesisへ一度に送信するデータバッチの最大サイズを指定します。サイズを調整することでデータ転送の効率と性能を細かく制御できます。「Batch Size」が「1」の場合、データレコードはバッチ化されず個別に送信されます。	`1`
Inflight Window	「インフライトキューリクエスト」とは、送信済みだが応答やアックをまだ受け取っていないリクエストのことです。この設定はSinkがKinesisと通信中に同時に存在可能なインフライトリクエストの最大数を制御します。 Request Modeが`asynchronous`の場合、このパラメータは特に重要です。同一MQTTクライアントからのメッセージを厳密に順序処理したい場合は、この値を`1`に設定してください。	`100`

Kubernetes

EMQX Operator

EMQX クラスター管理

API リファレンス

パスワード認証

監視

アクセスコントロール

統合

管理

Prometheus との統合

OpenTelemetry との統合

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

プラグインカタログ

JT/T 808 ゲートウェイ

GB/T 32960 ゲートウェイ

Volcengine RTC

シナリオ

GPT-Realtime

Amazon KinesisへのMQTTデータストリーム ​

動作原理 ​

特長とメリット ​

はじめる前に ​

前提条件 ​

Amazon Kinesis Data Streamsでのストリーム作成 ​

Amazon Kinesis Data Streamsのローカルエミュレーション ​

コネクターの作成 ​

Amazon Kinesis Sink付きルールの作成 ​

ルールのテスト ​

LocalStackでの確認方法 ​

詳細設定 ​

Amazon KinesisへのMQTTデータストリーム

動作原理

特長とメリット

はじめる前に

前提条件

Amazon Kinesis Data Streamsでのストリーム作成

Amazon Kinesis Data Streamsのローカルエミュレーション

コネクターの作成

Amazon Kinesis Sink付きルールの作成

ルールのテスト

LocalStackでの確認方法

詳細設定