$iceberg 集計ステージ

型の変換

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

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

ChangeStream のアーカイブ

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

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

2. $iceberg ステージでは、Amazon Web Services S3 バケットへの接続を確立し、iceberg-warehouse/ パスの myTable という名前のテーブルに書き込みます。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 ステージでは、Amazon Web Services S3 バケットへの接続を確立し、iceberg-warehouse/ パス内の myTable という名前の Apache Iceberg テーブルに書き込みます。cdc モードでは、各ドキュメントのstream.source.operationTypeメタデータ フィールドを読み取ることで、このステージで Apache Iceberg テーブルに対して実行する操作を決定します。

{
  "$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 Icebergターゲット

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

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

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

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

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

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

4. $iceberg ステージは、myData という名前の Amazon Web Services S3 バケットへの接続を確立し、iceberg-warehouse/ パス内の Apache Iceberg テーブルに書き込みます。ドキュメント メタデータから取得されたソース 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"
  }
}