$iceberg 集計ステージ

型の変換

Atlas Stream Processing は $iceberg ステージのテーブルに書き込む際に、BSON から Iceberg primitive types への型変換を実行します。

その他の BSON types はサポートされていません。Atlas Stream Processing は、サポートされていない BSON types のドキュメントをDLQ に送信します。

オブジェクト フィールドと配列フィールド

フィールドは、 schemaInference.modeAtlas Stream Processing が フィールドとobjectarray フィールドをApacheテーブルに書き込む方法を決定します。

• jsonモードでは、Atlas Stream Processingobject arraystringjsonは フィールドと フィールドを基本JSON文字列として直列化し、 列に書込みます。 はデフォルトのモードです。

• nestedモードでは、Atlas Stream Processing は フィールドと フィールドから Ops Managerobject arrayのネストされた型を推論します。

  • 各 objectフィールドはstruct 列になります。 Atlas Stream Processing は、オブジェクト内の対応するフィールドから、 構造体内の各フィールドの型を推論します。これは、ネストのすべてのレベルに適用されます。

  • 各 arrayフィールドはlist 列になります。 Atlas Stream Processing は、配列の null 以外の要素からリスト要素の型を推論します。プリミティブ要素はすべて同じ型である必要があります。要素がオブジェクトの場合、要素型はすべてのオブジェクトのフィールドを含む struct です。

モードは、ターゲット テーブルに一致する列がないフィールドにのみ適用されます。列がすでに存在する場合、Atlas Stream Processing はモードを無視し、 列タイプ に基づいてフィールドを書込みます。

• 列がstring の場合、Atlas Stream Processing はフィールドを基本JSON string として直列化します。

• 列が struct または list の場合、Atlas Stream Processing はそのネストされたタイプとしてフィールドを書込みます。

入力に新しい object または array フィールドが含まれている場合、Atlas Stream Processing はモードを使用して新しい列のタイプを推論します。

ChangeStream のアーカイブ

次の例は、Atlas データベースの初期コンテンツと変更ストリームを追加のみの形式で Apache Iceberg テーブルに書き込み、そのデータベースの操作履歴の耐久性のあるアーカイブを作成する方法を示しています。この集計には 2 つのステージがあります。

1. $source ステージでは、Atlas データベースとの接続を確立し、特定の db データベース内の orders コレクションをターゲットにします。これにより、プロセッサーの有効化時にデータベース内のドキュメントをキャプチャする最初の同期が可能になり、各チェンジストリーム イベントでフル ドキュメントがキャプチャされることが保証されます。

2. $icebergステージはAWS S3 バケットへの接続を確立し、myTable パス内のiceberg-warehouse/ という名前のテーブルに書き込みます。insert 操作のみを指定することで、追加専用のログスタイルの書込みフローが保証されます。

{
  "$source": {
    "connectionName": "atlas1",
    "db": "db",
    "coll": "orders",
    "initialSync": {
      "enable": true
    },
    "config": {
      "fullDocument": "required"
    }
  },
  "$iceberg": {
    "connectionName": "myS3Connection",
    "bucket": "myData",
    "path": "iceberg-warehouse/",
    "tableName": "myTable",
    "mode": "insert"
  }
}

コレクションのミラーリング

次の例は、Atlas コレクションを完全に Apache Iceberg テーブルにミラーリングする方法を示しています。

集計を定義する前に、次の変数を設定します。

const isDeleteExpr = {$eq: [{$meta: "stream.source.operationType"}, "delete"]};

次の集計では、Atlas ソース コレクションへの変更と同期して、Apache Iceberg テーブルエントリを追加、更新、削除します。これには 4 つのステージがあります。

1. $source ステージでは、Atlas データベースとの接続を確立し、特定の db データベース内の orders コレクションをターゲットにします。これにより、プロセッサーの有効化時にデータベース内のドキュメントをキャプチャする最初の同期が可能になり、各チェンジストリーム イベントでフル ドキュメントがキャプチャされることが保証されます。

2. $match ステージはoperationTypeでフィルターし、有効な操作タイプ宣言を持つドキュメントのみを処理します。

3. $replaceRoot ステージは、操作の種類に応じてドキュメントルートを変更します。

   • 削除操作の場合、ドキュメントのルートをドキュメントのキーに変更します。これにより、ドキュメントが削除されたことを示すレコードが作成されますが、そのコンテンツはそれ以降のプロセシングから除外されます。

   • その他のすべての操作については、ドキュメント ルートを fullDocument に変更し、ドキュメントのコンテンツをさらにプロセシングするために渡し、changestream メタデータを除外します。

4. $icebergステージはAWS S3 バケットへの接続を確立し、 myTableiceberg-warehouse/パスにある という名前のApache Tiger テーブルに書き込みます。cdc モードでは、このステージは各ドキュメントの メタデータフィールドから読み取ることで、 Apache Ticketrstream.source.operationType テーブルに対して実行する操作を決定します。

{
  "$source": {
    "connectionName": "atlas1",
    "db": "db",
    "coll": "orders",
    "initialSync": {
      "enable": true
    },
    "config": {
      "fullDocument": "required"
    }
  },
  "$match": {
    "operationType": {
      "$in": ["insert", "update", "delete", "replace"]
    }
  },
  "$replaceRoot": {
    "newRoot": {
      "$cond": {
        "if": isDeleteExpr,
        "then": "$documentKey",
        "else": "$fullDocument"
      }
    }
  }
  "$iceberg": {
    "connectionName": "myS3Connection",
    "bucket": "myData",
    "path": "iceberg-warehouse/",
    "tableName": "myTable",
    "mode": "cdc"
  }
}

マルチコレクションソースからマルチテーブルApacheターゲットへ

次の例では、Atlas Stream Processingは動的式を使用して、ドキュメントを様々な出力先に動的にルーティングします。

1. $source ステージでは、Atlas データベースとの接続を確立し、特に db データベース内のコレクション a、b、cを対象とします。これにより、プロセッサーの有効化時にデータベース内のドキュメントをキャプチャする最初の同期が可能になり、各チェンジストリーム イベントでフル ドキュメントがキャプチャされることが保証されます。

2. $match ステージは、operationType が "insert"、"update"、"delete"、"replace" のいずれかであるドキュメントをフィルターします。

3. $replaceRoot ステージは、操作の種類に応じてドキュメントルートを変更します。

   • 削除操作の場合、ドキュメントのルートをドキュメントのキーに変更します。これにより、ドキュメントが削除されたことを示すレコードが作成されますが、そのコンテンツはそれ以降のプロセシングから除外されます。

   • その他のすべての操作については、ドキュメント ルートを fullDocument に変更し、ドキュメントのコンテンツをさらにプロセシングするために渡し、changestream メタデータを除外します。

4. $icebergステージは という名前のAWS S3 バケットへの接続を確立し、myData iceberg-warehouse/パスのApacheテーブルに書き込みます。ドキュメントメタデータから検索されたソースcollection の名前に基づいて、テーブルの名前が決定されます。また、ドキュメントメタデータに従って実行する操作も決定します。

{
  "$source": {
    "connectionName": "atlas1",
    "db": "db",
    "coll": ["a", "b", "c"],
    "initialSync": {
      "enable": true
    },
    "config": {
      "fullDocument": "required"
    }
  }
},
{
  "$match": {
    "operationType": {
      "$in": ["insert", "update", "delete", "replace"]
    }
  }
},
{
  "$replaceRoot": {
    "newRoot": {
      "$cond": {
        "if": {
          "$eq": [{ "$meta": "stream.source.operationType" }, "delete"]
        },
        "then": "$documentKey",
        "else": "$fullDocument"
      }
    }
  }
},
{
  "$iceberg": {
    "connectionName": "myS3Connection",
    "databaseName": "iceberg-db",
    "bucket": "myData",
    "path": "iceberg-warehouse/",
    "tableName": { "$meta": "stream.source.ns.coll" },
    "mode": "cdc"
  }
}