Изменение потока в Azure DocumentDB

Потоки изменений — это поток изменений базы данных в режиме реального времени, который передается из базы данных в приложение. Эта функция позволяет создавать реактивные приложения, подписавшись на изменения базы данных, устраняя необходимость непрерывного опроса для обнаружения изменений.

Замечание

Azure DocumentDB в настоящее время поддерживает Поток изменений в кластерах с одним сегментом. Мы активно работаем над поддержкой потока изменений для кластеров с несколькими сегментами в ближайшем будущем.

Azure DocumentDB применяет ограничение журнала, управляемого системой, 400 МБ для потоков изменений. Клиенты должны непрерывно обрабатывать потоки изменений, чтобы не пропустить события исторических изменений. Потоки изменений используют системные журналы с ограничением размера 400 МБ. Если маркер возобновления ссылается на событие изменения, которое уже удалено из журнала, поток изменений нельзя возобновить с этого момента, и его необходимо перезапустить. Мы планируем предварительный показ расширенной поддержки в ближайшем будущем.

Включение потоков изменений

Эту функцию можно включить или отключить с помощью Azure CLI или шаблона ARM. Поддержка портала будет добавлена в ближайшее время.

Действия по включению потоков изменений в кластере с помощью ИНТЕРФЕЙСА командной строки

1. Вход в Azure CLI

az login

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

az resource show --ids "/subscriptions/<sub id>/resourceGroups/<resource group name>/providers/Microsoft.DocumentDB/mongoClusters/<resource name of your Azure DocumentDB cluster>" --api-version 2024-10-01-preview

3. Отправьте запрос PATCH, чтобы включить эту функцию.

az resource patch --ids "/subscriptions/<subscription_id>/resourceGroups/<resource_group_name>/providers/Microsoft.DocumentDB/mongoClusters/<cluster-name>" --api-version 2024-10-01-preview --properties "{\"previewFeatures\": [ \"ChangeStreams\"]}"

4. Проверьте результат:
   • Убедитесь, что нагрузка в ответе включает "previewFeatures": ["ChangeStreams"].
   • Если вы столкнулись с ошибкой "Потоки изменений не поддерживаются в этом кластере", создайте запрос на поддержку.

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

В этом примере код инициирует поток изменений в exampleCollection коллекции, непрерывно отслеживая любые изменения. При обнаружении изменения он извлекает событие изменения и выводит его в формате JSON.

JavaScript

// Open a change stream
const changeStream = db.exampleCollection.watch();

// Listen for changes
while (changeStream.hasNext()) 
    {
        const change = changeStream.next();
        printjson(change);
    }

Java

package com.example;
 
import com.mongodb.MongoClientSettings;
import com.mongodb.client.MongoClients;
import com.mongodb.client.MongoClient;
import com.mongodb.client.MongoDatabase;
import com.mongodb.client.MongoCollection;
import com.mongodb.client.MongoCursor;
import com.mongodb.client.model.changestream.ChangeStreamDocument;
 
import org.bson.BsonDocument;
import org.bson.Document;
import org.bson.json.JsonWriterSettings;
import org.json.JSONObject;
 
public class ChangeStreamExample {
    public static void main(String[] args) {
        MongoClient mongoClient = MongoClients.create("mongodb://<username>:<pwd>@<clustername>.global.mongocluster.cosmos.azure.com/?tls=true");
        MongoDatabase database = mongoClient.getDatabase("test");
        MongoCollection<Document> collection = database.getCollection("test");
 
        // Open a change stream
        MongoCursor<ChangeStreamDocument<Document>> cursor = collection.watch().iterator();
 
        // Listen for changes
        try {
            while (cursor.hasNext()) {
                ChangeStreamDocument<Document> change = cursor.next();
 
                // Create a JSONObject from the fields of the ChangeStreamDocument
                JSONObject json = new JSONObject();
 
                // Add fields to the JSON object
                json.put("operationType", change.getOperationType().getValue());
                json.put("namespace", change.getNamespace().getFullName());
 
                if (change.getDocumentKey() != null) {
                    json.put("documentKey", change.getDocumentKey().toJson().toString());
                }
 
                if (change.getResumeToken() != null) {
                    json.put("resumeToken", change.getResumeToken().toJson());
                }
 
                if (change.getFullDocument() != null) {
                    json.put("fullDocument", change.getFullDocument().toJson());
                }
 
                if (change.getWallTime() != null) {
                    json.put("wallTime", change.getWallTime().toString());
                }
 
                // Pretty-print the JSON object with 4-space indent
                System.out.println(json.toString(4)); 
            }
        } 
        catch (Exception e) {
            System.err.println("Error processing change stream: " + e);
        } 
        finally {
            cursor.close();
            mongoClient.close();
        }
    }
}

Important

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

cursor = db.exampleCollection.watch(resume_after=resume_token)

Мониторинг изменений базы данных с помощью потока изменений

Давайте рассмотрим выходные данные потока изменений с помощью примера.

Insert

В этом событии потока изменений мы видим, что новая запись была inserted в exampleCollection коллекции в cs базе данных, а сведения о событии включают полное содержимое только что добавленного документа.

{
  "_id": { "_data": "AeARBpQ/AAAA" }, // "resume_token"
  "operationType": "insert",
  "fullDocument": {
    "_id": { "$oid": "66e6f63e6f49ecaabf794958" },
    "employee_id": "17986",
    "name": "John",
    "position": "Software Engineer",
    "department": "IT",
    "rating": 4
  },
  "ns": { "db": "cs", "coll": "exampleCollection" },
  "documentKey": { "_id": { "$oid": "66e6f63e6f49ecaabf794958" } }
}

Update

В этом событии обновления position и rating для Джона изменены. Поток изменений отражает изменение update в коллекции exampleCollection с состоянием документа после обновления.

{
  "_id": { "_data": "AWACAKM/AAAA" },
  "operationType": "update",
  "fullDocument": {
    "_id": { "$oid": "66e6f63e6f49ecaabf794958" },
    "employee_id": "17986",
    "name": "John",
    "position": "SSE",
    "department": "IT",
    "rating": 5
  },
  "ns": {
    "db": "cs", "coll": "exampleCollection"
  },
  "documentKey": {
    "_id": { "$oid": "66e6f63e6f49ecaabf794958" }
  }
}

Удалить

Событие потока изменений указывает, что документ был deleted из коллекции exampleCollection в базе данных cs. Событие фиксирует уникальный идентификатор удаленного документа.

{
  "_id": { "_data": "ASgBAJs/AAAA" },
  "operationType": "delete",
  "ns": { "db": "cs", "coll": "exampleCollection" },
  "documentKey": { "_id": { "$oid": "66e6f63e6f49ecaabf794958" } }
}

Персонализация данных в потоке изменений

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

• $addFields
• $match
• $project
• $set
• $unset

Ограничения

• События потока изменений в кластере с несколькими сегментами пока не поддерживаются.
• Курсоры потоков изменений необходимо повторно инициализировать после события отказа в текущем состоянии.
• Update событие пока не поддерживает описание обновления.
• pre-image — неподдерживаемый параметр.
• $changestream как вложенный конвейер другого этапа еще не поддерживается.

Связанный контент

• Интегрируйте потоки изменений с Kafka (с помощью соединителя Debezium).