/ /

`$iceberg` 聚合阶段

定义

$iceberg 阶段在Connection Registry 中指定与 AWS S3 存储桶的连接，您可以将数据写入Apache Iceberg 表。

展示位置

$iceberg 必须是其所在的任何管道的最后阶段。每个管道只能使用一个 $iceberg 阶段。

语法

$iceberg 管道阶段采用以下原型形式：

{
  "$iceberg": {
    "connectionName": "<registered-connection>",
    "bucket": "<target-bucket>",
    "databaseName": "<database>",
    "tableName": "<string>" | <expression>,
    "path": "<key-prefix>",
    "region": "<target-region>",
    "mode": "cdc" | "insert",
    "idFieldName": "<field-name>",
    "partitionedBy": {
      "<column-name>": "<partition-transform>",
      . . .
    },
    "catalog": {
      "type": "hadoop" | "glue"
    }
  }
}

$iceberg 阶段采用包含以下字段的文档：

字段	类型	必要性	说明
`connectionName`	字符串	必需	用于读取和写入的 Amazon Web Services S3 连接名称。这必须与连接注册表中的连接名称匹配。
`bucket`	字符串	必需	包含目标 Apache Iceberg 数据库的 S 存储桶的名称。3
`databaseName`	字符串	必需	包含目标表的 Apache Iceberg 数据库的名称。
`tableName`	字符串 \|表达式	必需	目标 Apache Iceberg 表的名称。必须是 string 或计算结果为 string 的表达式。使用表达式实现每个文档的动态路由。
`path`	字符串	必需	路径前缀密钥，指向 Apache Iceberg 数据库。
`region`	字符串	可选的	存储桶的 Amazon Web Services 区域。在 Amazon Web Services 上运行的流处理器所必需。
`mode`	字符串	Optional	确定每个输入文档要执行的操作的策略。 `"cdc"` 导致 Atlas Stream Processing 通过读取 `stream.source.operationType` 元数据字段来确定操作类型。 `"insert"` 导致 Atlas Stream Processing 将每个文档作为新行添加到目标表中，忽略 `stream.source.operationType` 元数据字段中的操作类型声明。默认值为 `cdc`。
`idFieldName`	字符串	Optional	在 `cdc` 模式下用作行键的字段和列名称。默认值为 `"_id"`。
`partitionedBy`	文档	Optional	分区规格。如果您未设置此字段，`$iceberg` 将为 `idFieldName` 列设置默认分区转换。必须是包含一个或多个键值对的文档。每个键必须是要对其执行分区转换的列的名称，每个值必须是要使用的分区转换。第一个分区转换必须对应 `idFieldName`。给定字段的分区转换值必须为以下值之一： `"identity"` `"year"` `"month"` `"day"` `"hour"` `{ truncate: int }` `{ bucket: int }` 有关 Apache Iceberg 分区转换的更多信息，请参阅 Apache Iceberg 文档。
`catalog`	文档	Optional	定义要使用的 Iceberg 目录的文档。必须是包含 `type` 字段的文档，其值为 `"hadoop"` 或 `"glue"`。

行为

如果使用 $iceberg 阶段，则该阶段必须是流处理器中的最后阶段。

Atlas Stream Processing 仅支持 SP10、SP30 和 SP50 流处理器的 $iceberg 阶段。您的处理器层级决定动态路由支持的最大表数：

层级	最大表
SP 10	5
SP 30	10
SP 50	50

$iceberg 阶段从 Stream Processor 的输出数据模式推断生成的 Apache Iceberg 表的模式。随着 Atlas Stream Processing 观察到流中的新字段，表模式也会随之演进。

如果您指定的表尚不存在，则 Apache Iceberg 会在收到第一条针对该表的消息时创建该表。

Atlas Stream Processing 为输出到 Apache Iceberg 表提供至少一次处理。

您可以使用动态表达式作为 tableName 字段的值。通过使用动态表达式捕获文档特定值，您可以根据这些值将输入文档路由到不同的表中。表达式的结果必须为字符串。有关示例，请参阅动态路由。要了解详情，请参阅表达式操作符。

如果您使用动态表达式指定主题，但 Atlas Stream Processing 无法评估给定消息的表达式，则 Atlas Stream Processing 会将该消息发送到死信队列（DLQ）（如果已配置）并处理后续消息。如果未配置死信队列（DLQ），则 Atlas Stream Processing 会完全跳过该消息并处理后续消息。

类型转换

Atlas Stream Processing 在向 $iceberg 阶段中的表写入时，会将类型从 BSON 转换为 Iceberg 原始类型。

BSON	Apache Iceberg 原始	详情
`string`	`string`
`int`	`int`
`long`	`long`
`double`	`double`
`bool`	`boolean`
`ObjectId`	`string`	十六进制编码
`UUID`	`string`	字符化 UUID
`BinData`	`binary`	不适用于 UUID
`date`	`timestamptz`	以微秒为单位的 UTC 时间
`timestamp`	`timestamptz`	UTC 时间，测量到微秒
`object`	`string`	序列化为 Basic JSON string
`array`	`string`	序列化为 Basic JSON string

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

示例

以下示例说明了 $iceberg 阶段的各种应用。

存档变更流

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

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

下面的聚合会添加、更新和删除 Apache Iceberg 表目录，以与 Atlas 源集合的更改保持同步。它有四个阶段：

$source 阶段与 Atlas 数据库建立连接，具体针对 db 数据库中的 orders 集合。它可以在处理器激活时启用初始同步以捕获数据库中的文档，并确保每个变更流事件都能捕获完整的文档。
$match 阶段对 operationType 进行**过滤器**，以便仅处理具有有效操作类型声明的文档。
$replaceRoot 阶段会根据操作类型更改文档根。
- 对于删除操作，它会将文档根更改为文档的密钥。这会导致生成一条记录，表明文档已被删除，但会将其内容从进一步处理中排除。
- 对于所有其他操作，它会将文档根更改为 fullDocument，传递文档内容以供进一步处理，同时排除变更流元数据。
$iceberg 阶段建立与 AWS S3 存储桶的连接，并将数据写入名为 myTable 的 Apache Iceberg 表的 iceberg-warehouse/ 路径中。在 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 使用动态表达式将文档动态路由至各种输出目标。

$source 阶段与 Atlas 数据库建立连接，具体针对位于 db 数据库中的集合 a、b 和 c。它可以在处理器激活时启用初始同步以捕获数据库中的文档，并确保每个变更流事件都能捕获完整的文档。
$match 阶段会过滤 operationType 为 "insert"、"update"、"delete" 或 "replace" 之一的文档。
$replaceRoot 阶段会根据操作类型更改文档根。
- 对于删除操作，它会将文档根更改为文档的密钥。这会导致生成一条记录，表明文档已被删除，但会将其内容从进一步处理中排除。
- 对于所有其他操作，它会将文档根更改为 fullDocument，传递文档内容以供进一步处理，同时排除变更流元数据。
$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"
  }
}

后退

$externalFunction

来年

安全性

$iceberg 聚合阶段

定义