$iceberg 聚合阶段

类型转换

在 $iceberg 阶段写入表时， Atlas Stream Processing会执行从BSON开始的类型转换。 to Iceberg primitive types

不支持其他 BSON types。Atlas Stream Processing 会将包含不支持的 BSON types 的文档发送到 DLQ。

对象和数组字段

schemaInference.mode字段确定Atlas Stream Processing如何将 object 和 array 字段写入Apache Iceberg 表：

• 在 json模式下， Atlas Stream Processing将 object 和 array 字段序列化为基本JSON字符串，并将它们写入 string 列。 json 是默认模式。

• 在 nested模式下， Atlas Stream Processing会从 object 和 array 字段推断 Iceberg 嵌套类型：

  • 每个 object字段都成为 struct 列。 Atlas Stream Processing会根据对象中的相应字段推断结构体中每个字段的类型。这适用于每个嵌套级别。

  • 每个 array字段都成为 list 列。 Atlas Stream Processing从大量的非空元素推断列表元素类型。原始元素必须都是同一类型。如果元素是对象，则元素类型是包含所有对象字段的 struct。

该模式仅适用于目标表中没有匹配列的字段。如果列已经存在， Atlas Stream Processing会忽略模式并根据列类型写入字段：

• 如果列是 string， Atlas Stream Processing会将该字段序列化为 Basic JSON字符串。

• 如果列是 struct 或 list， Atlas Stream Processing会将该字段写入该嵌套类型。

如果输入包含新的 object 或 array 字段， Atlas Stream Processing会使用该模式来推断新列的类型。

存档变更流

以下示例演示了如何以仅追加的方式写入Atlas数据库的初始内容和变更流写入Apache Iceberg 表，从而创建该数据库操作历史记录的持久性存档。此聚合有两个阶段：

1. $source 阶段与 Atlas 数据库建立连接，具体针对 db 数据库中的 orders 集合。它可以在处理器激活时启用初始同步以捕获数据库中的文档，并确保每个变更流事件都能捕获完整的文档。

2. $iceberg 阶段建立与 AWS 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 表条目。它分为四个阶段：

1. $source 阶段与 Atlas 数据库建立连接，具体针对 db 数据库中的 orders 集合。它可以在处理器激活时启用初始同步以捕获数据库中的文档，并确保每个变更流事件都能捕获完整的文档。

2. $match 阶段对 operationType 进行**过滤器**，以便仅处理具有有效操作类型声明的文档。

3. $replaceRoot 阶段会根据操作类型更改文档根。

   • 对于删除操作，它会将文档根更改为文档的密钥。这会导致生成一条记录，表明文档已被删除，但会将其内容从进一步处理中排除。

   • 对于所有其他操作，它会将文档根更改为 fullDocument，传递文档内容以供进一步处理，同时排除变更流元数据。

4. $iceberg 阶段建立与 AWS S3 存储桶的连接，通过读取每个文档的 stream.source.operationType元数据字段来写入Apache Iceberg 表。 table named myTable in the iceberg-warehouse/ path. In cdc mode, this stage determines the operation to perform against the 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 Target

在以下示例中，Atlas Stream Processing 使用动态表达式将文档动态路由至各种输出目标。

1. $source 阶段与 Atlas 数据库建立连接，具体针对位于 db 数据库中的集合 a、b 和 c。它可以在处理器激活时启用初始同步以捕获数据库中的文档，并确保每个变更流事件都能捕获完整的文档。

2. $match 阶段会过滤 operationType 为 "insert"、"update"、"delete" 或 "replace" 之一的文档。

3. $replaceRoot 阶段会根据操作类型更改文档根。

   • 对于删除操作，它会将文档根更改为文档的密钥。这会导致生成一条记录，表明文档已被删除，但会将其内容从进一步处理中排除。

   • 对于所有其他操作，它会将文档根更改为 fullDocument，传递文档内容以供进一步处理，同时排除变更流元数据。

4. $iceberg 阶段建立与名为 myData 的 AWS 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"
  }
}