データ プロセッサ パイプラインで最後の既知の値を使用する

データ プロセッサ パイプライン内の最後の既知の値 (LKV) ステージを使用して、データの最新の完全なレコードを保守します。 LKV ステージでは、パイプライン内のメッセージに関するキーと値のペアの最新の値を追跡します。 その後このステージで、追跡対象の LKV 値を使用してメッセージをエンリッチできます。 最後の既知の値の追跡とエンリッチメントは、次のものに依存するダウンストリーム プロセスにとって重要です。

• 特定の timestamp での複数の時系列のデータ ポイント。
• 特定のキーの値が常にあるペイロード。

データ処理パイプラインでは、LKV ステージはオプションのステージです。 LKV ステージを使用すると、次のことができます。

• パイプラインに複数の LKV ステージを追加します。 各 LKV ステージで、複数の値を追跡できます。
• 保存された LKV 値を使用してメッセージをエンリッチし、データが完全で包括的な状態に保たれるようにします。
• 受信メッセージからの最新の値を使用して LKV が自動的に更新される状態を維持します。
• 論理パーティションごとに別個に LKV を追跡します。 LKV ステージは、各論理パーティション内で独立して動作します。
• 追跡対象の LKV ごとに期限切れ日時を構成して、有効な状態に保たれる期間を管理します。 このように管理すると、古い値を使用してメッセージがエンリッチされないようにするのに役立ちます。

LKV ステージでは、時系列のデータ整合性が維持されます。 このステージでは、古い timestamp のメッセージで、新しい timestamp のメッセージを使用して LKV がオーバーライドされたり置き換えられたりしません。

LKV ステージは、追跡対象の最後の既知の値を使用して受信メッセージをエンリッチします。 これらのエンリッチされた値は以前に記録されたデータを表しており、必ずしも現在のリアルタイムの値であるとは限りません。 このビヘイビアーが想定されているデータ処理に合っていることを確認してください。

前提条件

集計パイプライン ステージを構成して使用するには、オプションのデータ プロセッサ コンポーネントを含むデータ プロセッサのデプロイされたインスタンスが必要です。

ステージの構成

LKV ステージの JSON 構成で、ステージの詳細を定義します。 ステージを作成するには、フォーム ベースの UI を操作するか、[詳細設定] タブで JSON 構成を指定します。

timestamp パスを含めると、LKV が記録された時点を正確に把握し、透明性と追跡可能性を高めることができます。

inputPath と outputPath が等しい

送信メッセージは、実際のメッセージ値か、または追跡対象のキーがメッセージ ペイロード内にない場合は LKV の値です。 受信値が優先され、このステージで LKV を使用してオーバーライドされることはありません。 メッセージ値が LKV 値かどうかを識別するには、timestamp パスを使用します。 timestamp パスは、メッセージ内の値が追跡対象 LKV の場合のみ送信メッセージ内に含まれます。

inputPath と outputPath が等しくない

すべての受信メッセージについて、このステージで LKV が outputPath に書き込まれます。 この構成は、後続のメッセージ ペイロード内の値の違いを追跡するのに使用します。

サンプル構成

次の例は、LKV ステージのサンプル メッセージを示しています。このメッセージは 10:02 に着信し、ペイロードに追跡対象の LKV 値 .payload.temperature が含まれています。

{ 
  { 
    "systemProperties":{ 
        "partitionKey":"pump", 
        "partitionId":5, 
        "timestamp":"2023-01-11T10:02:07Z" 
    }, 
    "qos":1, 
    "topic":"/assets/pump/#" 
    }, 
    "payload":{ 
        "humidity": 10, 
        "temperature":250, 
        "pressure":30, 
        "runningState": true 
    } 
} 

LKV 構成:

追跡対象の LKV 値は次のとおりです。

• .payload.temperature は 250 です。
• LKV の timestamp は 2023-01-11T10:02:07Z です

11:05 に着信し、ペイロードに温度のプロパティがないメッセージの場合、次のように LKV ステージは追跡対象の値を使用してメッセージをエンリッチします。

11:05 の時点の LKV ステージへの入力例:

{ 
    "systemProperties":{ 
        "partitionKey":"pump", 
        "partitionId":5, 
        "timestamp":"2023-01-11T11:05:00Z" 
    }, 
    "qos":1, 
    "topic":"/assets/pump/#" 
    }, 
    "payload":{ 
        "runningState": true 
    } 
} 

11:05 の時点の LKV ステージからの出力例:

{ 
    "systemProperties":{ 
        "partitionKey":"pump", 
        "partitionId":5, 
        "timestamp":"2023-01-11T11:05:00Z" 
    }, 
    "qos":1, 
    "topic":"/assets/pump/#" 
    }, 
    "payload":{ 
        "lkvtemperature":250, 
        "lkvtemperature_timestamp"":"2023-01-11T10:02:07Z" 
        "runningState": true 
    } 
} 

関連するコンテンツ

• パイプライン内データの集計
• パイプライン内データのエンリッチメント
• パイプライン内のデータをフィルター処理する
• パイプラインから gRPC エンドポイントを呼び出す
• パイプラインから HTTP エンドポイントを呼び出す
• パイプライン内データを変換する