IoTフリート監視パイプラインの構築#

#目的

このチュートリアルでは、Data Platform上の完全なリアルタイムパイプラインを説明します。カスタムアクションからシミュレートされたIoTセンサーのテレメトリをストリーミングし、Apache Kafkaに送信し、Lakehouse Managerのテーブルに取り込み、Supersetダッシュボードでフリートの健全性をリアルタイムで可視化します。

最後に、デバイスの健全性、環境のトレンド、異常、ダウンタイムを表示する動作するダッシュボードを作成できます。これは、スクリプト化された故障シナリオによって駆動されるため、データは常にストーリーを語ります。

#はじめに

#必要条件

このチュートリアルに従うには、以下が必要です：

• OVHcloud Data Platformのアカウント

• OVHcloud Data Platformのプロジェクト

• OVHcloud Data Platformのプロジェクトにアクセスするための権限

• OVHcloud Data Platformのプロジェクトにアクセスするための認証情報

• OVHcloud Data Platform プロジェクトで、Data Processing Engine、Lakehouse Manager、および Superset/Trino 設定が利用可能です。

• 書き込み可能な Apache Kafka ブローカー。このチュートリアルでは、クライアント証明書（mTLS）認証を使用する OVH 管理 Kafka サービスを使用しますが、接続設定を適応させれば、どのブローカーでもこのアプローチが機能します。

• ブローカー上に作成された iot_readings という名前の Kafka トピック（トピックの自動作成は無効になっていることが多いです）。

最初に、最初のGetting StartedチュートリアルとApache Kafkaからのストリームデータチュートリアルを完了することをおすすめします。このガイドでは、プラットフォームの主要コンポーネントに慣れていることを前提としています。

#構築するもの

flowchart TD
    P[DPE Custom action: producer] --> K[Apache Kafka topic: iot_readings]
    K --> C[Kafka source connector]
    C --> L[DPE Load action: Always-up]
    L --> T[Lakehouse Manager table: iot_readings]
    T --> TR[Trino]
    TR --> S[Superset dashboards]

    classDef process fill:#E6F3FA,stroke:#0089C0,color:#0a2a36;
    classDef store fill:#0089C0,stroke:#0089C0,color:#fff;
    class P,C,L,TR,S process;
    class K,T store;

#シナリオ

プロデューサは、3つの地域にわたる4つのサイトに設置された環境センサー（温度、湿度、CO₂、PM2.5）のフリートをシミュレートします。各デバイスごとの状態マシンが、現実的な故障ライフサイクルを駆動するため、データは決して平坦ではありません。

stateDiagram-v2
    [*] --> OK
    OK --> DEGRADED: scripted trigger or battery < 15%
    DEGRADED --> FAULT: after 120s
    FAULT --> OFFLINE: after 90s, stops emitting
    OFFLINE --> OK: repaired after 60s

2つのデバイスは固定スケジュールで故障するようにスクリプトされています。そのため、各実行は常に完全なストーリーを示します。読み取り値は、ランダムノイズを含む圧縮された10分の「日常サイクル」に従います。そのため、トレンドは信頼性があります。デバイスがOFFLINEになると、完全にデータを送信しなくなります。これは後でデータの実際のギャップとして表示されます。

#ステップ1: Kafkaの資格情報をバケットに保存する

カスタムアクションのソースに証明書やパスワードを貼り付けるのではなく、Lakehouse Manager バケットに保存し、実行時に取得するようにしてください。カスタムアクションはプロジェクトに対して自己認証を行うため、追加の認証情報なしでバケットを読み取ることができます。また、コード内に機密情報を保持する必要もありません。

1. Lakehouse Manager で、iot-demo-certs という名前のバケットを作成します。
2. そのバケットに、3つの mTLS ファイルをアップロードします。

プロダクサは、起動時にSDKのバケットコネクタを使用してこれらをダウンロードします。

#ステップ2: プロデューサーカスタムアクションを作成する

プロダクサーをダウンロードし、カスタムアクションとして追加します。

1. Data Processing Engine > Actions > New > Custom を選択します。
2. ダウンロードした producer.py を直接アップロードします。
3. アクションの情報パネルでエントリ関数を customfunc に設定します。
4. アクションの Python Requirements に kafka-python を追加します。
5. 実行モードを Always-up に設定します。プロデューサは連続的にループするため、Serverless mode ではタイムアウトします。

#調整可能なパラメータ

すべての設定はファイルの先頭にあります。最も変更する可能性が高い設定は以下の通りです。

#コードのハイライト

ファイル全体を読む必要はありませんが、いくつかの部分は知っておく価値があります。

接続ブロック。 ここでブローカーのエンドポイントと認証方法を設定し、実行が有限かどうかを選択します。

KAFKA_MODE = "ssl"
TOPIC = "iot_readings"
BOOTSTRAP = "<your-kafka-bootstrap-host>:<port>"
MAX_RUNTIME_SECS = None  # set seconds for a finite, SUCCESS run

バケットからの認証情報。 証明書は実行時に取得され、ハードコーディングされることはありません。したがって、アクション内に機密情報は含まれません。

CERT_SOURCE = "bucket"
CERT_BUCKET = "iot-demo-certs"
# resolve_certs() downloads the CA, client cert, and key to /tmp via the SDK

フリート。 シミュレーションのサイズを変更するには、サイトを変更するか、サイトごとの数を変更します。

SITES = [
    ("paris-dc1",  "EU-W", "indoor-air"),
    ("london-dc2", "EU-W", "indoor-air"),
    # ...
]
DEVICES_PER_SITE = 3   # 4 sites x 3 -> 12 devices

失敗シナリオ。 2つのデバイスが固定スケジュールで故障し、これらの期間がOKからDEGRADED、FAULT、OFFLINEへのストーリーのペースを決めます。

PHASE_DEGRADED, PHASE_FAULT, PHASE_OFFLINE, PHASE_REPAIRED_COOLDOWN = 120, 90, 60, 90

アクションを実行します。各メッセージには、安定した型を持つ十二のフィールドがすべて含まれているため、データが流れ始めるとすぐにスキーマが完了します。最初のスクリプト化されたデバイスは約十五秒後に劣化しますので、DEGRADED と FAULT のサンプルはすぐに現れます。アクションログでハートビート行を確認してください: sent ~N messages; fleet: {...}。

#ステップ3: ソースコネクタを使用してKafkaに接続する

接続 > ソース > Apache Kafka に移動し、接続を作成します。ブローカーのエンドポイントと iot_readings トピックを指定してください。

mTLS ブローカーの場合は、コネクタの SSL フィールド で認証を行います。CA 証明書、クライアント証明書、クライアントキーをアップロードし、ユーザー名とパスワードは空白のままにしてください。

#ステップ4：メタデータを抽出し、テーブルを作成する

データが流れ始めたら、Analyzerを開き、メタデータを抽出します。

16つの属性が表示されます：プロデューサからの12フィールドに加え、Kafkaが追加した4つのエンベロープ列（timestamp、date、offset_r、partition）です。シンプルな追加のみのテーブルの場合、4つのエンベロープ列はすべてスキップできます。partitionとoffset_rは、アップサートまたは重複排除のセマンティクスが必要な場合のみ保持してください。

プロデューサはtsをISO-8601形式の文字列として出力します。文字列として推論された場合、テーブルでTimestamp型に設定するか、Trinoで解析する（Good to knowを参照）かのいずれかを選択できます。結果のスキーマからLakehouse Managerテーブル iot_readingsを構築します。

完全な列のリスト：

#ステップ5：ストリームをテーブルに読み込む

Data Processing EngineでLoadアクションを作成し、iot_readingsトピックをiot_readingsテーブルにマッピングします。Always-upモードで実行して、ライブストリームを継続的に消費するようにします。

正常に完了したLoadアクションは、メタデータの更新を自動的に実行するため、テーブルの行数は自動的に更新されます。このストリーミングセットアップでは、アクションはAlways-upで実行され、手動で停止するかタイムアウトするまで実行を続けるため、自動更新が実行されない場合があります。その場合は、Update Metadataアクションを実行して、Lakehouse Managerの行数が取り込まれたデータを反映するようにします。

#ステップ6：Supersetからテーブルをクエリする

Trinoを介してSupersetをテーブルに接続します。Supersetをまだデプロイしていない場合は、Deploy Apache Supersetに従ってください。

Supersetで、SQLAlchemy URIがtrino://<user>@<trino-host>:<port>/<catalog>形式のデータベース接続を追加します。チャートを構築する前に接続をテストします。これはエンドツーエンドのセットアップで最も頻繁に破損するステップです。

tsが文字列として到着した場合は、Trinoでfrom_iso8601_timestamp(ts)を使用して解析し、TIMESTAMP(3) WITH TIME ZONEを返します。解析された列を公開する仮想データセットを作成し、主時系列列としてマークします。これにより、時系列チャートのX軸として利用可能になります。

#ステップ7：ダッシュボードを構築する

完成したダッシュボードは、フリート全体のシングルスクリーンビューです。

実用的なパネルのセット：

仮想データセット全体と各パネルの正確なTrino SQLおよび構成については、次のページに従ってください：

Build the Superset dashboards

Supersetのいくつかのヒント：

• バーチャートでは、カテゴリをX軸に配置し、Dimensionsボックスを空白にします。Dimensionsは、各バーをサブシリーズに分割するためだけです。
• 閾値線（例：バッテリー警告レベル）は、時系列チャートでアノテーションレイヤーを通じて利用可能です。カテゴリバーチャートでは、代わりに昇順に並べ替えます。テーブルでは、条件付き書式設定を使用します。
• CO₂値（約480）は、温度（約21）とPM2.5（約9）を共通軸で圧倒します。CO₂を別途チャート化するか、二次軸を使用して、小さなメトリクスが読み取り可能なままにします。
• ダウンタイムテーブルはフリートが健全な場合は空になります。これは正しい動作です。フィルタなしで構築し、すべてのデバイスをリスト表示し、条件付き書式設定で古い行を強調表示して、パネルが壊れたように見えないようにします。

#Good to know

• プロデューサはデザイン上終了しません。 これはAlways-upモードの無限ループです。アクションのデフォルトタイムアウトは約2時間です。手動で停止すると、「停止」と表示され、成功ではなく表示されます。これは自然な終了がないためです。クリーンな成功実行を得るには、MAX_RUNTIME_SECSを秒単位の数値に設定します。ループはフラッシュされ、閉じられ、合計がログに記録され、終了します。連続的なライブダッシュボードにするには、Noneのままにします。
• Update Metadataで行数を更新します。 正常に終了したLoadアクションは、テーブルメタデータを自動的に更新します。このストリーミングセットアップでは、アクションはAlways-upで実行され、終了する代わりに停止またはタイムアウトするため、ステップ5で説明したように、行数が取り込まれた内容を反映するにはUpdate Metadataアクションを実行する必要がある場合があります。
• 時系列用にtsを解析します。 列が文字列の場合は、from_iso8601_timestamp(ts)で解析し、Supersetで時系列列としてマークします。
• 解析不能な行に対処します。 テストメッセージがtsが有効なタイムスタンプでない行を残した場合、from_iso8601_timestampはINVALID_FUNCTION_ARGUMENTをスローします。try(from_iso8601_timestamp(ts))で解析をラップして、悪い行がnullに解決するようにするか、行を削除します。

#構築したもの

現在、完全なストリーミングパイプラインを持っています：カスタムアクションがシミュレートされたテレメトリを生成し、Kafkaがそれを輸送し、コネクタとLoadアクションがそれをLakehouse Managerに取り込み、SupersetがTrinoを介してそれをライブで可視化します。スクリプトされた故障が進行するにつれて、ダッシュボードは全体のストーリーを通過します：健全なフリート、劣化して故障するデバイス、ダウンタイムテーブルとトレンドのギャップとして現れるダウンタイムのウィンドウ、そして最後にデバイスを正常に戻す修理です。

ここから、別のデバイス次元テーブルでモデルを拡張し、さらに多くのメトリクスを追加するか、プロデューサを変更して独自の実際のセンサーデータを再生することができます。

#さらに詳しく

当社のソリューションを実装するためのトレーニングや技術的なアシスタンスが必要な場合は、営業担当者にお問い合わせください、またはこのリンクをクリックして見積もりを取得し、当社のプロフェッショナルサービスの専門家にプロジェクトのカスタム分析を依頼してください。

DiscordチャネルでData Platformを構築するチームと直接やり取りし、質問をしたり、フィードバックを送信したり、交流したりしてください。

OVHcloudサービスについてサポートが必要な場合は、ヘルプセンターでリクエストを作成してください。

ユーザーコミュニティに参加してください。