Ler dados em tempo real com fluxos de alterações

Os fluxos de alterações do Firestore com compatibilidade com o MongoDB permitem que os aplicativos acessem mudanças em tempo real (inserções, atualizações e exclusões) feitas em uma coleção ou em um banco de dados inteiro. Um fluxo de alterações ordena as atualizações por horário de modificação.

Os fluxos de alterações podem ser acessados pelas APIs compatíveis com o MongoDB e pelos drivers tradicionais do MongoDB. A implementação de fluxos de alterações do Firestore com compatibilidade com o MongoDB pode processar qualquer capacidade de gravações e leituras por meio de uma implementação exclusiva de particionamento automático em gravações e paralelismo de leitura. Isso permite criar cargas de trabalho de alta capacidade de processamento. Também é possível melhorar a infraestrutura de migração e sincronização de dados entre o Cloud Firestore e outras soluções de armazenamento.

Além da compatibilidade com os drivers do MongoDB, é possível usar o Cloud Firestore para ler fluxos de alterações em paralelo. Isso permite criar cargas de trabalho de leitura paralelas e de alta capacidade de processamento. Cada stream representa uma partição bem distribuída de resultados.

Os fluxos de alterações são compatíveis com os seguintes recursos:

  • Fluxos de alterações configuráveis com escopo de banco de dados ou coleção.
  • Um período de armazenamento para um fluxo de alterações especificado na criação. A retenção padrão é de 7 dias, e a mínima é de 1 dia. A retenção precisa ser um múltiplo de 1 dia, até no máximo 7 dias. A duração da retenção não pode ser alterada após a criação. Para mudar o período de retenção, é necessário remover e recriar o fluxo de alterações.
  • Eventos de mudança de delete, insert, update e drop que são observáveis usando db.collection.watch() e db.watch().
  • updateDescription.updatedFields contém diferenças de atualização.
  • Todas as opções fullDocument e fullDocumentBeforeChange.
    • Procurando atualizações no documento completo.
    • Pré-imagem do documento antes de ser substituído, atualizado ou excluído.
    • Imagem posterior do documento após a substituição ou atualização.
    • As imagens pré e pós-evento com mais de uma hora exigem a ativação da recuperação pontual (PITR).
  • Todas as opções de currículo, incluindo resumeAfter e startAfter.
  • Ao usar watch() para observar mudanças, é possível encadear estágios de agregação como $addFields, $match, $project, $replaceRoot, $replaceWith, $set e $unset.

Configurar fluxos de alterações

Para criar, descartar ou visualizar os fluxos de alterações de um banco de dados, use o console do Google Cloud.

Papéis e permissões

Para criar, excluir e listar fluxos de alterações, um principal precisa das permissões datastore.schemas.create, datastore.schemas.delete e datastore.schemas.list do Identity and Access Management (IAM), respectivamente.

O papel Administrador de índice do armazenamento de dados (roles/datastore.indexAdmin), por exemplo, concede essas permissões.

Criar um fluxo de alterações

Antes de abrir um cursor de fluxo de alterações correspondente, é preciso criar um fluxo de alterações. Não é possível ativar o fluxo de alterações automaticamente na criação de coleções ou bancos de dados.

Para criar um fluxo de alterações, use o console do Google Cloud.

  1. No Console do Google Cloud, acesse a página Bancos de Dados.

    Acessar "Bancos de dados"

  2. Na lista, selecione um banco de dados do Firestore com compatibilidade com o MongoDB. O painel do Firestore Studio é aberto.
  3. No painel Explorador, encontre o nó Fluxos de alterações, clique em Mais ações e selecione Criar fluxo de alterações.
  4. Insira um nome, escopo e período de retenção exclusivos para o fluxo de alterações e clique em Salvar.

Ver fluxos de alterações

É possível ver detalhes sobre os fluxos de alterações no console do Google Cloud.

  1. No Console do Google Cloud, acesse a página Bancos de Dados.

    Acessar "Bancos de dados"

  2. Na lista, selecione um banco de dados do Firestore com compatibilidade com o MongoDB. O painel do Firestore Studio é aberto.
  3. No painel Explorador, encontre o nó Fluxos de alterações.
  4. Para abrir ou fechar o nó, clique em Alternar nó.

Excluir um fluxo de alterações

Para excluir um fluxo de alterações, use o console do Google Cloud.

  1. No Console do Google Cloud, acesse a página Bancos de Dados.

    Acessar "Bancos de dados"

  2. Na lista, selecione um banco de dados do Firestore com compatibilidade com o MongoDB. O painel do Firestore Studio é aberto.
  3. No painel Explorador, encontre o nó Fluxos de alterações.
  4. Para abrir ou fechar o nó, clique em Alternar nó.
  5. No Explorer, localize o fluxo de alterações que você quer excluir.
  6. Clique em Mais ações e selecione Excluir fluxo de alterações.
  7. Na caixa de diálogo, insira o nome do fluxo de alterações para confirmar a exclusão e clique em Excluir.

Abrir ou retomar um cursor de fluxo de alterações

Os exemplos a seguir demonstram como criar, retomar e configurar um cursor de fluxo de alterações.

Antes de criar um cursor de fluxo de alterações, é necessário criar um fluxo de alterações para o banco de dados ou a coleção.

Criar um cursor de fluxo de alterações

Para criar um novo cursor de fluxo de alterações, use o método watch nos drivers do MongoDB. Para detectar todas as mudanças em um banco de dados, crie um fluxo de alterações com escopo de banco de dados e chame o método watch no objeto db.

let cursor = db.watch()

Para criar um cursor com escopo para uma coleção, primeiro é necessário criar um fluxo de alterações para essa coleção. Em seguida, chame o método watch na coleção correspondente.

let cursor = db.my_collection.watch()

Agora que você criou um cursor de fluxo de alterações, pode começar o streaming. Por exemplo, se você inserir um documento e chamar tryNext no cursor, a mudança vai aparecer no fluxo de alterações.

let doc = db.my_collection.insertOne({value: "hello world"})
console.log(cursor.tryNext())

Se você atualizar e excluir o documento, essas mudanças vão aparecer no fluxo de alterações:

db.my_collection.updateOne({"_id": doc.insertedId}, {$set: {value: "hello world!"}})
db.my_collection.deleteOne({"_id": doc.insertedId}})

// Prints the update event
console.log(cursor.tryNext())

// Prints the delete event
console.log(cursor.tryNext())

Retomar um fluxo de alterações

Para retomar um fluxo de alterações, use as opções resumeAfter ou startAfter. Para determinar onde retomar o registro de alterações de resumeAfter e startAfter, use um token de retomada.

// Create a cursor and add one event to the change stream.
let cursor = db.my_collection.watch();
db.my_collection.insertOne({value: "hello world"});
let event = cursor.tryNext();

// Get the resume token from the event.
let resumeToken = event._id;

// Add a new event to the change stream.
db.my_collection.insertOne({value: "foobar"});

// Create a new cursor by using the resume token as a starting point.
let newCursor = db.my_collection.watch({resumeAfter: resumeToken})

// Log the change event containing the "foobar" value.
console.log(newCursor.tryNext())

Para usar startAfter:

// Start after the resume token.
let startAfterCursor = db.my_collection.watch({startAfter: resumeToken})

Incluir imagens pré e pós em atualizações e exclusões

Se necessário, é possível incluir imagens de documentos antes e depois em eventos de alteração de atualização e exclusão. A disponibilidade da imagem está sujeita à janela de recuperação pontual (PITR, na sigla em inglês). Para ler imagens de documentos com mais de uma hora, é necessário ativar a PITR.

Os fluxos de alterações aproveitam a janela de PITR para fornecer uma visualização do documento antes e depois do evento de alteração. Por padrão, os eventos de atualização contêm um campo updateDescription, que é o delta dos campos modificados pela operação de atualização.

Para incluir as imagens pré e pós em um evento de alteração, especifique as opções fullDocumentBeforeChange e fullDocument na consulta do fluxo de alterações.

let cursor = db.my_collection.watch({
  "fullDocument": "required",
  "fullDocumentBeforeChange": "required"
})

Se a consulta tentar ler um documento fora da janela de retenção da PITR ou se a PITR não estiver ativada, o valor required vai gerar uma mensagem de erro do lado do servidor.

Como alternativa para gerar um erro, use o valor whenAvailable para retornar um valor null se as imagens não estiverem mais disponíveis.

let cursor = db.my_collection.watch({
  "fullDocument": "whenAvailable",
  "fullDocumentBeforeChange": "whenAvailable"
})

Incluir a imagem atual nas atualizações

Por padrão, os eventos de atualização contêm um campo updateDescription, que é o delta dos campos modificados pela operação de atualização. Para pesquisar a versão mais atual de todo o documento, use o valor updateLookup na opção fullDocument.

Esse recurso não requer PITR e realiza uma pesquisa pelo documento.

let cursor = db.my_collection.watch({
  "fullDocument": "updateLookup",
})

Leituras paralelas

Para aumentar a capacidade, use a opção firestoreWorkerConfig para dividir uma consulta de fluxo de alterações entre vários workers. Cada worker é responsável por exibir as alterações de um conjunto distinto de documentos. É preciso criar um cursor paralelo usando uma consulta runCommand ou aggregate.

Por exemplo, é possível distribuir um fluxo de alterações entre três workers da seguinte maneira:

let cursor1 = db.my_collection.aggregate([{
    "$changeStream": {
        "firestoreWorkerConfig": {numWorkers: 3, workerId: 0 }}
  }]);

let cursor2 = db.my_collection.aggregate([{
    "$changeStream": {
        "firestoreWorkerConfig": {numWorkers: 3, workerId: 1 }}
  }]);

let cursor3 = db.my_collection.aggregate([{
    "$changeStream": {
        "firestoreWorkerConfig": {numWorkers: 3, workerId: 2 }}
  }]);

Fluxos de alterações e backups

Nem a configuração nem os dados do fluxo de alterações estão disponíveis nas operações de restauração de backup. Se você restaurar um banco de dados com fluxos de alterações, recrie esses fluxos no banco de dados de destino para abrir cursores para esse banco de dados.

Faturamento

Diferenças de comportamento

A seção a seguir descreve as diferenças nos fluxos de alterações entre o Firestore com compatibilidade com o MongoDB e o MongoDB.

updateDescription

updateDescription é um documento em um evento update que descreve os campos que foram atualizados ou removidos pela operação de atualização. No Cloud Firestore, as diferenças importantes são:

  • Em updateDescription, os campos truncatedArrays e disambiguatedPaths não são preenchidos.
  • updateDescription.updatedFields representam uma diferença canônica entre as imagens pré e pós de um documento antes e depois da aplicação de uma mutação.

Considere o seguinte estado inicial de um documento:

db.my_collection.insertOne({
  _id: 1,
  root: {
    array: [{a: 1}, {b: 2}, {c: 3}]
  }
})

Cenário 1: mudar apenas o primeiro elemento da matriz.

Nesse cenário, o comportamento do Cloud Firestore corresponde ao do MongoDB.

db.my_collection.updateOne(
  {_id: 1},
  {'$set': {"root.array.0.a": 100}}
)

{
  updatedFields: {"root.array.0.a": 100},
  removedFields: []
}

Cenário 2: substituir por uma matriz inteira

Nesse cenário, a operação atualiza apenas o primeiro campo da matriz, mas substitui toda a matriz.

A diferença de atualização de Cloud Firestore não diferencia esses dois cenários e retorna o mesmo updateDescription.updatedFields para ambos:

db.my_collection.updateOne(
  {_id: 1},
  {'$set': {"root.array": [{a: 100}, {b: 2}, {c: 3}]}}
)

// In other implementations, updatedFields reflects the mutation itself
{
  updatedFields: {
    "root.array": [{a: 100}, {b: 2}, {c: 3}]
  },
  removedFields: []
}

// Firestore updatedFields is the diff between the before and after versions of the document
{
  updatedFields: {"root.array.0.a": 100},
  removedFields: []
}

A seguir