$iceberg Aggregation Stage

Type Conversion

Atlas Stream Processing performs type conversion from BSON to Iceberg primitive types when writing to tables in the $iceberg stage.

Other BSON types are unsupported. Atlas Stream Processing sends documents with unsupported BSON types to the DLQ.

Archiving a Changestream

The following example demonstrates how to write the initial contents and changestream of an Atlas database to an Apache Iceberg table in an append-only fashion, creating a durable archive of that database's operational history. This aggregation has two stages:

1. The $source stage establishes a connection with the Atlas database, specifically targeting the orders collection in the db database. It enables initial sync to capture documents in the database at the time of processor activation, and ensures that the full document is captured by each changestream event.

2. The $iceberg stage establishes a connection to the AWS S3 bucket, writing to a table named myTable in the iceberg-warehouse/ path. By specifying only insert operations, it ensures an append-only, log-style write flow.

{
  "$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"
  }
}

Mirroring a Collection

The following example demonstrates how to mirror an Atlas collection in its entirety to an Apache Iceberg table.

Before defining the aggregation, set the following variable:

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

The following aggregation adds, updates, and deletes Apache Iceberg table entries in sync with changes to the Atlas source collection. It has four stages:

1. The $source stage establishes a connection with the Atlas database, specifically targeting the orders collection in the db database. It enables initial sync to capture documents in the database at the time of processor activation, and ensures that the full document is captured by each changestream event.

2. The $match stage filters on operationType so that only documents with valid operation type declarations are processed.

3. The $replaceRoot stage changes the document root depending on the type of operation.

   • For delete operations, it changes the document root to the document's key. This results in a record that the document was deleted but excludes its content from further processing.

   • For all other operations, it changes the document root to the fullDocument, passing the content of the document along for further processing while excluding changestream metadata.

4. The $iceberg stage establishes a connection to the AWS S3 bucket, writing to an Apache Iceberg table named myTable in the iceberg-warehouse/ path. In cdc mode, this stage determines the operation to perform against the Apache Iceberg table by reading from the stream.source.operationType metadata field of each document.

{
  "$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"
  }
}

From Multi-Collection Source to Multi-Table Apache Iceberg Target

In the following example, Atlas Stream Processing uses dynamic expressions to dynamically route documents to various output targets.

1. The $source stage establishes a connection with the Atlas database, specifically targeting the collections a, b, and c in the db database. It enables initial sync to capture documents in the database at the time of processor activation, and ensures that the full document is captured by each changestream event.

2. The $match stage filters for documents whose operationType is one of "insert", "update", "delete", or "replace".

3. The $replaceRoot stage changes the document root depending on the type of operation.

   • For delete operations, it changes the document root to the document's key. This results in a record that the document was deleted but excludes its content from further processing.

   • For all other operations, it changes the document root to the fullDocument, passing the content of the document along for further processing while excluding changestream metadata.

4. The $iceberg stage establishes a connection to the AWS S3 bucket named myData, writing to an Apache Iceberg table in the iceberg-warehouse/ path. It determines the name of the table according to the name of the source collection as retrieved from document metadata. It also determines the operation to perform according to document metadata.

{
  "$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"
  }
}