RedshiftへのMQTTデータ取り込み

Amazon Redshift は、ペタバイト規模のクラウドデータウェアハウスであり、高性能な分析を目的としたフルマネージドサービスです。PostgreSQLをベースにし、オンライン分析処理（OLAP）に最適化されているため、複雑なクエリの実行や大規模なデータ分析を高速に行えます。EMQXはAmazon Redshiftと直接連携し、IoTデバイスからのMQTTテレメトリをほぼリアルタイムで取り込み、保存することが可能です。

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

動作概要

EMQXのRedshiftデータ統合は組み込み機能であり、MQTTベースのIoTデータストリームをAmazon Redshiftの分散型PostgreSQL互換データウェアハウスに直接取り込みます。EMQXの組み込みルールエンジンを利用することで、複雑なカスタムコードを書かずにIoTデータをRedshiftにストリーミングし、大規模な分析処理が可能です。

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

EMQX Integration Redshift

RedshiftへのMQTTデータ取り込みの流れは以下の通りです：

IoTデバイスがEMQXに接続：IoTデバイスがMQTTプロトコルを介して正常に接続されると、オンラインイベントがトリガーされます。イベントにはデバイスID、送信元IPアドレスなどの情報が含まれます。
メッセージのパブリッシュと受信：デバイスは特定のトピックにテレメトリやステータスデータをパブリッシュします。EMQXがこれらのメッセージを受信すると、ルールエンジン内でマッチング処理が開始されます。
ルールエンジンによるメッセージ処理：EMQXのルールエンジンは、トピックやメッセージ内容に基づいて定義されたルールにマッチさせてイベントやメッセージを処理します。処理内容にはデータ変換（例：JSONからSQL用フォーマットへの変換）、フィルタリング、コンテキスト情報によるデータ強化などが含まれ、データベース挿入前に行われます。
Redshiftへの書き込み：マッチしたルールはSQLベースの取り込みをRedshiftに対してトリガーします。SQLテンプレートを用いて、処理済みデータのフィールドをRedshiftのテーブルおよびカラムにマッピングします。高スループットの取り込みには、Amazon S3からのCOPYコマンドやRedshift Streaming Ingestionを活用し、カラムナーストアに効率的にロードします。RedshiftのクエリオプティマイザとMPP（Massively Parallel Processing）実行エンジンにより、データは即座に分析クエリに利用可能となります。

イベントおよびメッセージデータがRedshiftに書き込まれた後は、以下のような活用が可能です：

Amazon QuickSight、Grafana、TableauなどのツールとRedshiftを接続し、IoTメトリクスやトレンドを追跡するダッシュボードを構築。
RedshiftデータをAWSの分析およびAI/MLサービス（例：Amazon SageMaker）と連携し、異常検知やデバイス挙動の予測を実施。
Redshiftの並列クエリ実行により、大規模なIoTデータセットに対して集計、結合、時系列分析を実行し、過去データとほぼリアルタイムのインサイトを提供。

特長と利点

Redshiftとのデータ統合により、以下の特長とメリットをビジネスにもたらします：

柔軟なイベント処理：EMQXのルールエンジンを活用し、Redshiftはデバイスのライフサイクルイベント（接続、切断、ステータス変化）を低レイテンシで保存・処理可能です。RedshiftのMPPクエリエンジンと組み合わせることで、障害検知、異常検知、長期利用傾向の迅速な集計・分析が行えます。
メッセージ変換：メッセージはEMQXルールで広範に処理・変換されてからRedshiftに書き込まれるため、保存データは分析に最適化された状態となります。これによりクエリの複雑さが軽減され、下流処理が効率化されます。
SQLテンプレートによる柔軟なデータ操作：EMQXのSQLテンプレートマッピングを通じて、構造化されたIoTデータをRedshiftのテーブル・カラムに挿入可能です。RedshiftはPostgreSQL互換SQL、JSON用のSUPER型などの半構造化データ型、クエリ最適化のための高度なインデックスをサポートします。カラムナーストレージ、データ圧縮、ゾーンマップにより、大規模データセットのスキャン時間を短縮しクエリを高速化します。
ビジネスプロセスの統合：RedshiftはAWSエコシステムとシームレスに統合されており、IoTデータをAmazon QuickSightなどのBIツール、AWS GlueやAWS Data Pipelineなどの分析サービス、Amazon SageMakerなどのAI/MLサービスに接続可能です。
高度な地理空間機能：RedshiftはGEOMETRYおよびGEOGRAPHY型を通じて地理空間データ型と関数をサポートし、ジオフェンシング、位置情報分析、ルート最適化を実現します。EMQXのリアルタイム取り込みと組み合わせることで、資産追跡、車両監視、位置ベースのイベントトリガーをほぼリアルタイムに行えます。
組み込みのメトリクスと監視：EMQXは各Redshiftシンクのランタイムメトリクスを提供し、RedshiftはAmazon CloudWatchと連携してクラスターのパフォーマンス、クエリ実行メトリクス、ストレージ使用状況を監視可能です。これにより、取り込みから分析までのエンドツーエンドの可観測性を確保します。

はじめる前に

このセクションでは、Redshift統合の作成を開始する前に必要な準備について説明します。Redshiftクラスターの作成、データベースおよびデータテーブルの作成方法を含みます。

前提条件

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

Amazon Redshiftでのデータベースおよびテーブル作成

EMQXでRedshiftコネクターを設定する前に、Amazon Redshiftクラスター（またはServerlessワークグループ）が稼働していること、そしてIoTデータを格納するスキーマが準備されていることを確認してください。

Redshiftクラスターまたはワークグループをデプロイします。Amazon Redshiftクラスター作成ガイドに従い環境を起動してください。
データベースユーザーの認証情報を設定します。初期クラスター作成時に、管理者ユーザー（通常はadminuser）の資格情報を指定します。
もしくは、Redshift SQLを使ってEMQX専用のデータベースユーザーを作成します。このユーザーには接続、テーブル作成、読み書きの権限が必要です。例：
sql
```
CREATE USER emqx_user PASSWORD 'YourStrongPassword1';
```
詳細はRedshift入門ガイドおよびユーザーガイドを参照してください。
後でEMQXのコネクター設定に使用するため、ユーザー名（emqx_user）とパスワードを控えておいてください。
任意のPostgreSQL互換クライアント（psql、SQL Workbench/J、DBeaverなど）を使用し、ホスト名、ポート、既存のデータベース名（例：デフォルトのdev）、ユーザー名、パスワードでRedshiftエンドポイントに接続します。
接続後、EMQXからのIoTデータ受け入れ先となるemqx_dataデータベースを作成します。
sql
```
CREATE DATABASE emqx_data;
```

emqx_dataデータベースに接続し、MQTTメッセージおよびクライアントイベントデータを格納するための2つのテーブルを作成します。

クライアントID、トピック、ペイロード、作成時刻を保存するデータテーブルt_mqtt_msgを以下のSQLで作成します：

sql

CREATE TABLE t_mqtt_msg (
  id BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
  msgid VARCHAR(64),
  sender VARCHAR(64),
  topic  VARCHAR(255),
  qos    INTEGER,
  retain INTEGER,
  -- ペイロードがJSONの場合はSUPER型を検討、そうでなければ大きめのVARCHARを使用
  payload SUPER,
  arrived TIMESTAMPTZ
);

クライアントのオンライン/オフラインイベントをタイムスタンプ付きで保存するemqx_client_eventsテーブルを以下のSQLで作成します：

sql

CREATE TABLE emqx_client_events (
  id BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
  clientid  VARCHAR(255),
  event     VARCHAR(255),
  created_at TIMESTAMPTZ DEFAULT CURRENT_TIMESTAMP
);

Redshiftコネクターの作成

Redshiftシンクを追加する前に、EMQXでRedshiftコネクターを作成する必要があります。コネクターはEMQXがAmazon RedshiftクラスターまたはServerlessワークグループに接続する方法を定義します。

注意

Amazon Redshift Serverlessを使用している場合、コネクターが作成され接続が確立されると、データが書き込まれていなくても課金が発生する可能性があります。未使用のコネクターは削除するか、リソースを一時停止して予期せぬコストを回避してください。

EMQXダッシュボードで、Integration -> Connector に移動します。
ページ右上の Create をクリックします。
Create Connector ページで Redshift を選択し、Next をクリックします。
コネクター名を入力します。名前は英数字で始まり、英数字、ハイフン、アンダースコアを含めることができます。例：my_redshift。
Redshift接続情報を入力します：
- Server Host：Redshiftエンドポイントのホスト名（例：redshift-cluster-1.abc123xyz.us-east-1.redshift.amazonaws.com）。AWS RedshiftコンソールのClustersまたはWorkgroupsページで確認可能です。
- Database Name：EMQXデータを格納する対象データベース。例：emqx_data。
- Username：データ挿入権限を持つデータベースユーザー名。例：emqx_user。
- Password：emqx_userのパスワード。
- Enable TLS：Redshift接続にSSL/TLS暗号化が必要な場合はオンにします（クラウドサービス接続では推奨）。詳細は外部リソースアクセスのTLSを参照。
詳細設定（任意）：接続プールサイズ、アイドルタイムアウト、リクエストタイムアウトなどの追加接続プロパティを設定可能です。詳細はシンクの機能を参照してください。
Test Connectivity をクリックし、EMQXが提供された設定でRedshiftクラスターに正常に接続できるか確認します。
Create をクリックしてコネクターを保存します。
作成後は以下のいずれかを選択できます：
- Back to Connector List をクリックして全コネクターを表示
- Create Rule をクリックして、このコネクターを使ったルールをすぐに作成し、Redshiftへのデータ転送を設定
詳細な例は以下を参照してください：
- メッセージ保存用Redshiftシンクのルール作成
- イベント記録用Redshiftシンクのルール作成

メッセージ保存用Redshiftシンクのルール作成

このセクションでは、ダッシュボードでソースMQTTトピックt/#からのメッセージを処理し、処理済みデータを設定済みのRedshiftシンクを介してt_mqtt_msgテーブルに保存するルールの作成方法を示します。

ダッシュボードの Integration -> Rules ページに移動します。
ページ右上の Create をクリックします。
ルールIDに my_rule を入力し、SQLエディターにルールを入力します。ここではトピックt/#のMQTTメッセージをRedshiftに保存するため、ルールのSELECT句でSQLテンプレート内で使用するすべての変数を含むフィールドを選択してください。例：
sql
```
SELECT
*
FROM
"t/#"
```
TIP
初心者の方は SQL Examples をクリックし、Enable Test を有効にしてSQLルールの学習とテストが可能です。
- Add Action ボタンをクリックし、ルールでトリガーされるアクションを定義します。このアクションにより、EMQXはルールで処理したデータをRedshiftに送信します。
Type of Action ドロップダウンからRedshiftを選択し、Action ドロップダウンはデフォルトの Create Action のままにするか、既存のRedshiftアクションを選択します。この例では新規シンクを作成しルールに追加します。
シンクの名前と説明をフォームに入力します。
Connector ドロップダウンから先ほど作成した my_redshift を選択します。新規コネクターを作成する場合はドロップダウン横のボタンをクリックしてください。設定パラメータはRedshiftコネクターの作成を参照。

SQL Template を設定します。以下のSQL文を使用してデータを挿入します。

注意：これはプリプロセス済みSQLのため、フィールドは引用符で囲まず、文末にセミコロンを付けないでください。

sql

INSERT INTO t_mqtt_msg (
    msgid,
    topic,
    qos,
    payload,
    arrived
)
VALUES (
    ${id},
    ${topic},
    ${qos},
    ${payload},
    timestamp 'epoch' + (${timestamp} :: bigint / 1000) * interval '1 second'
)

フォールバックアクション（任意）：メッセージ配信失敗時の信頼性向上のため、1つ以上のフォールバックアクションを定義可能です。詳細はフォールバックアクションを参照してください。
詳細設定（任意）：詳細はシンクの機能を参照してください。
Create をクリックする前に、Test Connectivity をクリックしてシンクがRedshiftサーバーに接続可能かテストできます。
Create ボタンをクリックし、シンクの設定を完了します。新しいシンクがAction Outputsに追加されます。
Create Rule ページで設定内容を確認し、Save をクリックしてルールを生成します。

ルール作成後、Integration -> Rules ページで新規ルールを確認でき、Action (Sink) タブで新規Redshiftシンクも確認可能です。

また、Integration -> Flow Designer でトポロジーを可視化し、トピックt/#のメッセージがルールmy_ruleで解析されRedshiftに書き込まれている様子を確認できます。

イベント記録用Redshiftシンクのルール作成

このセクションでは、クライアントのオンライン/オフライン状態を記録し、イベントデータを設定済みのRedshiftシンクを介してemqx_client_eventsテーブルに保存するルールの作成方法を示します。

手順はメッセージ保存用Redshiftシンクのルール作成とほぼ同様ですが、SQLテンプレートとSQLルールが異なります。

オンライン/オフライン状態記録用のSQLルールは以下の通りです：

sql

SELECT
  *
FROM
  "$events/client_connected", "$events/client_disconnected"

イベント記録用SQLテンプレートは以下の通りです：

注意：これはプリプロセス済みSQLのため、フィールドは引用符で囲まず、文末にセミコロンを付けないでください。

sql

INSERT INTO emqx_client_events(clientid, event, created_at) VALUES (
  ${clientid},
  ${event},
  TO_TIMESTAMP((${timestamp} :: bigint)/1000)
)

ルールのテスト

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

bash

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

2つのシンクの稼働状況を確認してください。メッセージ保存用シンクでは新規の受信メッセージと送信メッセージが1件ずつあるはずです。イベント記録用シンクでは2件のイベントレコードが確認できます。

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

bash

emqx_data=# select * from t_mqtt_msg;
 id |              msgid               | sender | topic | qos | retain |            payload
        |       arrived
----+----------------------------------+--------+-------+-----+--------+-------------------------------+---------------------
  1 | 0005F298A0F0AEE2F443000012DC0002 | emqx_c | t/1   |   0 |        | { "msg": "hello Redshift" } | 2023-01-19 07:10:32
(1 row)

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

bash

emqx_data=# select * from emqx_client_events;
 id | clientid |        event        |     created_at
----+----------+---------------------+---------------------
  3 | emqx_c   | client.connected    | 2023-01-19 07:10:32
  4 | emqx_c   | client.disconnected | 2023-01-19 07:10:32
(2 rows)

Kubernetes

EMQX Operator

EMQX クラスター管理

API リファレンス

パスワード認証

監視

アクセスコントロール

統合

管理

Prometheus との統合

OpenTelemetry との統合

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

プラグインカタログ

JT/T 808 ゲートウェイ

GB/T 32960 ゲートウェイ

Volcengine RTC

シナリオ

GPT-Realtime

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

動作概要 ​

特長と利点 ​

はじめる前に ​

前提条件 ​

Amazon Redshiftでのデータベースおよびテーブル作成 ​

Redshiftコネクターの作成 ​

メッセージ保存用Redshiftシンクのルール作成 ​

イベント記録用Redshiftシンクのルール作成 ​

ルールのテスト ​

RedshiftへのMQTTデータ取り込み

動作概要

特長と利点

はじめる前に

前提条件

Amazon Redshiftでのデータベースおよびテーブル作成

Redshiftコネクターの作成

メッセージ保存用Redshiftシンクのルール作成

イベント記録用Redshiftシンクのルール作成

ルールのテスト