$iceberg Estágio de agregação

conversão de tipo

O Atlas Stream Processing executa a conversão de tipo de BSON para tipos primitivos Iceberg ao gravar em tabelas no estágio $iceberg.

Outros BSON types não são suportados. O Atlas Stream Processing envia documentos com BSON types não suportados para o DLQ.

Arquivando um Changestream

O exemplo a seguir demonstra como gravar o conteúdo inicial e o changestream de um banco de dados Atlas em uma tabela Apache Iceberg de forma somente anexa, criando um arquivo durável do histórico operacional desse banco de dados. Esta agregação tem duas etapas:

1. O estágio $source estabelece uma conexão com o banco de dados Atlas, visando especificamente a coleção orders no banco de dados db. Ele permite que a sincronização inicial capture documentos no banco de dados no momento da ativação do processador e garante que o documento completo seja capturado por cada evento de changestream.

2. O estágio $iceberg estabelece uma conexão com o bucket AWS S3, gravando em uma tabela chamada myTable no caminho iceberg-warehouse/. Ao especificar apenas insert operações, ele garante um fluxo de gravação somente de acréscimo, estilo log.

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

Espelhando uma coleção

O exemplo a seguir demonstra como espelhar uma coleção do Atlas em sua totalidade para uma tabela Apache Iceberg.

Antes de definir a agregação, defina a seguinte variável:

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

A agregação a seguir adiciona, atualizar e excluir entradas de tabela Apache Iceberg em sincronização com as alterações na coleção de origem do Atlas. Tem quatro etapas:

1. O estágio $source estabelece uma conexão com o banco de dados Atlas, visando especificamente a coleção orders no banco de dados db. Ele permite que a sincronização inicial capture documentos no banco de dados no momento da ativação do processador e garante que o documento completo seja capturado por cada evento de changestream.

2. A etapa $match filtra operationType para que apenas documentos com declarações de tipo de operação válidas sejam processados.

3. A fase $replaceRoot altera a raiz do documento dependendo do tipo de operação.

   • Para operações de exclusão, ele altera a raiz do documento para a chave do documento. Isso resulta em um registro de que o documento foi excluído, mas exclui seu conteúdo em processamento posterior.

   • Para todas as outras operações, ele altera a raiz do documento para o fullDocument, passando o conteúdo do documento para em processamento adicional, excluindo metadados de fluxo de alteração.

4. O estágio $iceberg estabelece uma conexão com o bucket AWS S3, gravando em uma tabela Apache Iceberg chamada myTable no caminho iceberg-warehouse/. No modo cdc, este estágio determina a operação a ser executada na tabela Apache Iceberg lendo o campo de metadados stream.source.operationType de cada documento.

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

De fonte de várias coleções para destino de várias tabelas Apache Iceberg

No exemplo a seguir, o Atlas Stream Processing usa expressões dinâmicas para rotear dinamicamente documentos para vários destinos de saída.

1. O estágio $source estabelece uma conexão com o banco de dados Atlas, visando especificamente as coleções a, b e c no banco de dados db. Ele permite que a sincronização inicial capture documentos no banco de dados no momento da ativação do processador e garante que o documento completo seja capturado por cada evento de changestream.

2. O estágio $match filtra documentos cujo operationType é um de "insert", "update", "delete" ou "replace".

3. A fase $replaceRoot altera a raiz do documento dependendo do tipo de operação.

   • Para operações de exclusão, ele altera a raiz do documento para a chave do documento. Isso resulta em um registro de que o documento foi excluído, mas exclui seu conteúdo em processamento posterior.

   • Para todas as outras operações, ele altera a raiz do documento para o fullDocument, passando o conteúdo do documento para em processamento adicional, excluindo metadados de fluxo de alteração.

4. O estágio $iceberg estabelece uma conexão com o bucket AWS S3 chamado myData, gravando em uma tabela Apache Iceberg no caminho iceberg-warehouse/. Ele determina o nome da tabela de acordo com o nome da origem collection conforme recuperado dos metadados do documento. Ele também determina a operação a ser executada de acordo com os metadados do documento.

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