Как читать данные в реальном времени с помощью потоков изменений

Потоки изменений для Firestore с совместимостью с MongoDB позволяют приложениям получать доступ к изменениям в реальном времени (вставкам, обновлениям и удалениям), внесенным в коллекцию или во всю базу данных. Поток изменений упорядочивает обновления по времени изменения.

Доступ к Change Streams можно получить через API, совместимые с MongoDB, и традиционные драйверы MongoDB. Реализация потоков изменений в Firestore с совместимостью с MongoDB может обрабатывать любую пропускную способность операций записи и чтения благодаря уникальной реализации автоматического разбиения на разделы при записи и параллелизма чтения. Это позволяет создавать рабочие нагрузки с высокой пропускной способностью. Вы также можете улучшить инфраструктуру переноса и синхронизации данных между Cloud Firestore и другими решениями для хранения.

Помимо совместимости с драйверами MongoDB, вы можете использовать Cloud Firestore для параллельного чтения потоков изменений. Это позволяет создавать параллельные рабочие нагрузки чтения с высокой пропускной способностью. Каждый поток представляет собой хорошо распределенный раздел результатов.

Потоки изменений поддерживают следующие функции:

  • Настраиваемые потоки изменений с областью действия базы данных или коллекции.
  • Срок хранения для потока изменений, указанный при создании. По умолчанию срок хранения составляет 7 дней, а минимальный срок – 1 день. Срок хранения должен быть кратен одному дню и не может превышать семи дней. После создания срок хранения изменить нельзя. Чтобы изменить срок хранения, необходимо удалить и заново создать поток изменений.
  • delete, insert, update и drop изменяют события, которые можно отслеживать с помощью db.collection.watch() и db.watch().
  • updateDescription.updatedFields содержит различия в обновлениях.
  • Все возможности fullDocument и fullDocumentBeforeChange.
    • Поиск обновлений в полном документе.
    • Предварительное изображение документа до его замены, обновления или удаления.
    • Изображение документа после его замены или обновления.
    • Для изображений до и после, которые старше одного часа, требуется включить восстановление на определенный момент времени (PITR).
  • Все варианты резюме, включая resumeAfter и startAfter.
  • При использовании watch() для отслеживания изменений вы можете объединять этапы агрегирования, такие как $addFields, $match, $project, $replaceRoot, $replaceWith, $set и $unset.

Как настроить потоки изменений

Чтобы создать, удалить или посмотреть существующие потоки изменений для базы данных, используйте консоль Google Cloud.

Роли и разрешения

Чтобы создавать, удалять и просматривать потоки изменений, пользователю нужны разрешения Identity and Access Management (IAM) datastore.schemas.create, datastore.schemas.delete и datastore.schemas.list соответственно.

Например, эти разрешения предоставляет роль Администратор индексов хранилища данных (roles/datastore.indexAdmin).

Как создать поток изменений

Прежде чем открыть курсор потока изменений, необходимо создать поток изменений. Автоматическое включение потока изменений при создании коллекции или базы данных не поддерживается.

Чтобы создать поток изменений, используйте консоль Google Cloud.

  1. В консоли Google Cloud перейдите на страницу Databases (Базы данных).

    Перейти к базам данных

  2. Выберите из списка базу данных Firestore, совместимую с MongoDB. Откроется панель Firestore Studio.
  3. На панели Проводник найдите узел Потоки изменений, нажмите Ещё и выберите Создать поток изменений.
  4. Введите уникальное название потока изменений, область действия и срок хранения, а затем нажмите Сохранить.

Просмотр потоков изменений

Подробные сведения о потоках изменений можно посмотреть в консоли Google Cloud.

  1. В консоли Google Cloud перейдите на страницу Databases (Базы данных).

    Перейти к базам данных

  2. Выберите из списка базу данных Firestore, совместимую с MongoDB. Откроется панель Firestore Studio.
  3. На панели Проводник найдите узел Потоки изменений.
  4. Чтобы открыть или закрыть узел, нажмите Переключить узел.

Как удалить поток изменений

Чтобы удалить поток изменений, используйте консоль Google Cloud.

  1. В консоли Google Cloud перейдите на страницу Databases (Базы данных).

    Перейти к базам данных

  2. Выберите из списка базу данных Firestore, совместимую с MongoDB. Откроется панель Firestore Studio.
  3. На панели Проводник найдите узел Потоки изменений.
  4. Чтобы открыть или закрыть узел, нажмите Переключить узел.
  5. В Проводнике найдите поток изменений, который нужно удалить.
  6. Нажмите Ещё и выберите Удалить поток изменений.
  7. В диалоговом окне введите название потока изменений, чтобы подтвердить удаление, и нажмите Удалить.

Как открыть или возобновить курсор потока изменений

В следующих примерах показано, как создать, возобновить и настроить курсор потока изменений.

Прежде чем создавать курсор потока изменений, необходимо явным образом создать поток изменений для базы данных или коллекции.

Создание курсора потока изменений

Чтобы создать новый курсор потока изменений, используйте метод watch в драйверах MongoDB. Чтобы отслеживать все изменения в базе данных, создайте поток изменений на уровне базы данных и вызовите метод watch для объекта db.

let cursor = db.watch()

Чтобы создать курсор, ограниченный коллекцией, сначала необходимо создать поток изменений для этой коллекции. Затем вызовите метод watch для нужной коллекции.

let cursor = db.my_collection.watch()

Теперь, когда вы создали курсор потока изменений, можно начать потоковую передачу. Например, если вы вставите документ и вызовете tryNext для курсора, то увидите изменение в потоке изменений.

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

Если вы обновите и удалите документ, то увидите эти изменения в потоке изменений:

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())

Как возобновить поток изменений

Чтобы возобновить поток изменений, используйте параметр resumeAfter или startAfter. Чтобы определить, с какого места в истории изменений возобновить работу с resumeAfter и startAfter, используйте токен возобновления.

// 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())

Чтобы использовать startAfter, сделайте следующее:

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

Добавление изображений до и после в обновления и удаление

При необходимости вы можете включить в события изменения, связанные с обновлением и удалением, изображения документов до и после изменения. Доступность изображений зависит от окна восстановления на определенный момент времени (PITR). Чтобы читать изображения документов, созданные более часа назад, необходимо включить PITR.

Потоки изменений используют окно PITR, чтобы предоставить представление документа до и после заданного события изменения. По умолчанию события обновления содержат поле updateDescription, которое представляет собой дельту полей, измененных операцией обновления.

Чтобы включить в событие изменения изображения до и после, необходимо указать параметры fullDocumentBeforeChange и fullDocument в запросе потока изменений.

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

Если запрос пытается прочитать документ за пределами окна хранения PITR или если PITR не включен, значение required выдает сообщение об ошибке на стороне сервера.

Вместо того чтобы вызывать ошибку, вы можете использовать значение whenAvailable, чтобы возвращать значение null, если изображения больше не доступны.

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

Включить текущее изображение в обновления

По умолчанию события обновления содержат поле updateDescription, которое представляет собой разницу между полями, измененными в результате операции обновления. Чтобы найти самую последнюю версию всего документа, используйте значение updateLookup в параметре fullDocument.

Для этой функции не требуется PITR, и она выполняет поиск документа.

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

Параллельное чтение

Чтобы увеличить пропускную способность, можно использовать параметр firestoreWorkerConfig, который позволяет разделить запрос потока изменений между несколькими исполнителями. Каждый рабочий процесс отвечает за обработку изменений в определенном наборе документов. Для этого нужно создать параллельный курсор с помощью запроса runCommand или aggregate.

Например, вы можете распределить поток изменений между тремя рабочими процессами следующим образом:

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 }}
  }]);

Потоки изменений и резервные копии

При восстановлении из резервной копии недоступны ни конфигурация, ни данные потока изменений. Если вы восстанавливаете базу данных с помощью потоков изменений, вам необходимо воссоздать эти потоки изменений в целевой базе данных, чтобы открыть курсоры для этой базы данных.

Оплата

Различия в поведении

В следующем разделе описываются различия в потоках изменений между Firestore с совместимостью с MongoDB и MongoDB.

updateDescription

updateDescription – это документ в событии update, в котором описываются поля, обновленные или удаленные в результате операции обновления. В Cloud Firestore есть следующие отличия:

  • В updateDescription поля truncatedArrays и disambiguatedPaths не заполнены.
  • updateDescription.updatedFields представляет собой каноническую разницу между изображениями документа до и после применения изменения.

Рассмотрим следующий исходный документ:

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

Сценарий 1. Изменение только первого элемента массива.

В этом случае поведение Cloud Firestore соответствует MongoDB.

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

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

Сценарий 2. Перезапись всего массива

В этом случае операция обновляет только первое поле массива, но перезаписывает весь массив.

Разница между обновлениями Cloud Firestore не различает эти два сценария и возвращает одинаковое значение updateDescription.updatedFields для обоих:

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: []
}

Дальнейшие действия