$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.

Object and Array Fields

The schemaInference.mode field determines how Atlas Stream Processing writes object and array fields to Apache Iceberg tables:

• In json mode, Atlas Stream Processing serializes object and array fields as Basic JSON strings and writes them to string columns. json is the default mode.

• In nested mode, Atlas Stream Processing infers Iceberg nested types from object and array fields:

  • Each object field becomes a struct column. Atlas Stream Processing infers the type of each field in the struct from the corresponding field in the object. This applies at every level of nesting.

  • Each array field becomes a list column. Atlas Stream Processing infers the list element type from the non-null elements of the array. Primitive elements must all be the same type. If the elements are objects, the element type is a struct that contains the fields of all the objects.

The mode applies only to fields that don't have a matching column in the target table. If the column already exists, Atlas Stream Processing ignores the mode and writes the field based on the column type:

• If the column is a string, Atlas Stream Processing serializes the field as a Basic JSON string.

• If the column is a struct or list, Atlas Stream Processing writes the field as that nested type.

If the input contains new object or array fields, Atlas Stream Processing uses the mode to infer the types of the new columns.

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