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

Campos de objeto e array

O campo schemaInference.mode determina como o Atlas Stream Processing grava os campos object e array nas tabelas do Apache icebergue:

• No modojson, o Atlas Stream Processing serializa os campos object e array como strings JSON básicas e os grava em colunas string. json é o modo padrão.

• No modonested, o Atlas Stream Processing infere tipos aninhados de icebergue a partir de campos object e array:

  • Cada campo object se torna uma coluna struct . O Atlas Stream Processing infere o tipo de cada campo na estrutura a partir do campo correspondente no objeto. Isso se aplica a todos os níveis de aninhamento.

  • Cada campo array se torna uma coluna list . O Atlas Stream Processing infere o tipo de elemento da lista a partir dos elementos não nulos da array. Os elementos primitivos devem ser todos do mesmo tipo. Se os elementos forem objetos, o tipo de elemento é um struct que contém os campos de todos os objetos.

O modo se aplica apenas a campos que não têm uma coluna correspondente na tabela de destino. Se a coluna já existir, o Atlas Stream Processing ignorará o modo e gravará o campo baseado no tipo de coluna:

• Se a coluna for um string, o Atlas Stream Processing serializa o campo como uma string JSON básica.

• Se a coluna for struct ou list, o Atlas Stream Processing gravará o campo como esse tipo aninhado.

Se a entrada contiver novos campos object ou array, o Atlas Stream Processing usará o modo para inferir os tipos das novas colunas.

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 somente as operações insert, ele garante um fluxo de escrita somente para anexação e no estilo de registro.

{
  "$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 do AWS S3, gravando em uma tabela do Apache icebergue chamada myTable no caminho iceberg-warehouse/. No modocdc, esse estágio determina a operação a ser executada na tabela do Apache Geophyllum, 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"
  }
}

Da origem de várias coleções para o alvo Apache icebergue de várias tabelas

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 do AWS S3 chamado myData, gravando em uma tabela Apache icebergue 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"
  }
}