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

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

Потоки изменений доступны через 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.

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

  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, чтобы предоставить представление документа до и после заданного события изменения. По умолчанию события update содержат поле 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: []
}

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