データ プロセッサ パイプラインでのシリアル化と逆シリアル化の形式

データ プロセッサは、データに依存しないプラットフォームです。 データ プロセッサは、任意の形式でデータを取り込み、処理、書き出すことができます。

ただし、一部のパイプライン ステージで jq パス式を使用するには、データがパイプライン内の構造化形式である必要があります。 データを適切な構造化形式に変換するには、データの逆シリアル化が必要になる場合があります。

一部のパイプライン変換先またはステージからの呼び出しでは、場合によってはデータを特定の形式にする必要があります。 データを変換先に適した形式にシリアル化することが必要になる場合があります。

メッセージを逆シリアル化する

データ プロセッサは、データ ソース ステージと、パイプラインが外部データを読み取る呼び出しステージの両方で、さまざまな形式の逆シリアル化をネイティブにサポートします。

• ソース ステージでは、受信データを逆シリアル化できます。
• 呼び出しステージでは、API 応答を逆シリアル化できます。

次の場合は、受信データを逆シリアル化する必要がない場合があります。

• 逆シリアル化されたデータを必要とするステージを使用していない。
• メタデータのみを処理している。
• 受信データは既に、使用されているステージと一致する形式になっている。

次の表に、逆シリアル化がサポートされる形式と、対応するステージを示します。

メッセージをシリアル化する

データ プロセッサは、パイプラインが外部データを書き込む宛先と呼び出し先の両方のステージで、さまざまな形式へのシリアル化をネイティブにサポートします。

• 宛先ステージでは、送信データを適切な形式にシリアル化できます。
• 呼び出しステージでは、API 要求で送信されたデータをシリアル化できます。

Raw/JSON/MessagePack/CBOR データ形式

Raw は、データを逆シリアル化またはシリアル化する必要がない場合に使用するオプションです。 Raw は、逆シリアル化またはシリアル化が適用されないほとんどのステージで、既定値です。

シリアル化または逆シリアル化の構成は、Raw、JSON、MessagePack、CBOR 形式で共通です。 これらの形式の場合は、次の構成オプションを使用します。

データを逆シリアル化するには、次の構成オプションを使用します。

データをシリアル化するには、次の構成オプションを使用します。

次の例は、バッチ処理されていない JSON データをシリアル化または逆シリアル化するための構成を示しています。

{
    "format": {
        "type": "json",
        "path": ".payload"
    }
}

次の例は、ソース ステージで JSON データを逆シリアル化するか、バッチ処理された JSON データをシリアル化するための構成を示しています。

{
    "format": {
        "type": "json"
    }
}

プロトコル バッファーのデータ形式

プロトコル バッファー (protobuf) データを逆シリアル化するには、次の構成オプションを使用します。

protobuf データをシリアル化するには、次の構成オプションを使用します。

次の例は、バッチ処理されていない protobuf データをシリアル化または逆シリアル化するための構成を示しています。

{
    "format": {
        "type": "protobuf",
        "descriptor": "Zm9v..",
        "package": "package1",
        "message": "message1",
        "path": ".payload"
    }
}

次の例は、ソース ステージで protobuf データを逆シリアル化するか、バッチ処理された protobuf データをシリアル化するための構成を示しています。

{
    "format": {
        "type": "protobuf",
        "descriptor": "Zm9v...", // The full descriptor
        "package": "package1",
        "message": "message1"
    }
}

CSV データ形式

CSV データを逆シリアル化するには、次の構成オプションを使用します。

列配列内の各要素は、次のスキーマを持つオブジェクトです。

CSV データをシリアル化するには、次の構成オプションを使用します。

次の例は、バッチ処理されていない CSV データをシリアル化するための構成を示しています。

{
    "format": {
        "type": "csv",
        "header": true,
        "columns": [
            {
                "name": "assetId",
                "path": ".assetId"
            },
            {
                "name": "timestamp",
                "path": ".eventTime"
            },
            {
                "name": "temperature",
                // Path is optional, defaults to the name
            }
        ],
        "path": ".payload"
    }
}

次の例は、バッチ処理されている CSV データをシリアル化するための構成を示しています。 バッチ処理されているデータの最上位レベル path を省略します。

{
    "format": {
        "type": "csv",
        "header": true,
        "columns": [
            {
                "name": "assetId",
                "path": ".assetId"
            },
            {
                "name": "timestamp",
                "path": ".eventTime"
            },
            {
                "name": "temperature",
                // Path is optional, defaults to .temperature
            }
        ]
    }
}

次の例は、バッチ処理されていない CSV データを逆シリアル化するための構成を示しています。

{
    "format": {
        "type": "csv",
        "header": false,
        "columns": [
            {
                "name": "assetId",
                "type": "string",
                "path": ".assetId"
            },
            {
                "name": "timestamp",
                // Type is optional, defaults to string
                "path": ".eventTime"
            },
            {
                "name": "temperature",
                "type": "float"
                // Path is optional, defaults to .temperature
            }
        ],
        "path": ".payload"
    }
}

次の例は、ソース ステージでバッチ処理された CSV データを逆シリアル化するための構成を示しています。

{
    "format": {
        "type": "csv",
        "header": false,
        "columns": [
            {
                "name": "assetId",
                "type": "string",
                "path": ".assetId"
            },
            {
                "name": "timestamp",
                // Type is optional, defaults to string
                "path": ".eventTime"
            },
            {
                "name": "temperature",
                "type": "float",
                // Path is optional, defaults to .temperature
            }
        ]
    }
}

関連するコンテンツ

• データ プロセッサ メッセージ
• パーティション分割とは
• 構成パターンとは