Streamek módosítása az Azure Cosmos DB MongoDB-hez készült API-jában

A KÖVETKEZŐKRE VONATKOZIK: MongoDB

A változáscsatorna támogatása az Azure Cosmos DB MongoDB-hez készült API-jában a change streams API használatával érhető el. A change streams API használatával az alkalmazások lekérhetik a gyűjteményben vagy a szegmensben lévő elemeken végrehajtott módosításokat. Később további műveleteket is végrehajthat az eredmények alapján. A gyűjtemény elemeinek módosításait a módosítási idő sorrendjében rögzíti a rendszer, a rendezési sorrend pedig szegmenskulcsonként garantált.

Példák

Az alábbi példa bemutatja, hogyan kérhet le változásstreameket a gyűjtemény összes elemére. Ez a példa egy kurzort hoz létre az elemek beszúrása, frissítése vagy cseréjekor. A $match változásstreamek lekéréséhez szükség van a fázisra, $project a fázisra és fullDocument a beállításra. A módosítási streamekkel végzett törlési műveletek figyelése jelenleg nem támogatott. Áthidaló megoldásként hozzáadhat egy helyreállítható jelölőt a törölt elemekhez. Hozzáadhat például egy attribútumot a "törölt" nevű elemhez. Ha törölni szeretné az elemet, beállíthatja a "törölt" true értéket, és beállíthat egy TTL-t az elemen. Mivel a "törölt" true frissítés egy frissítés, ez a módosítás látható lesz a változásstreamben.

var cursor = db.coll.watch(
    [
        { $match: { "operationType": { $in: ["insert", "update", "replace"] } } },
        { $project: { "_id": 1, "fullDocument": 1, "ns": 1, "documentKey": 1 } }
    ],
    { fullDocument: "updateLookup" });

while (!cursor.isExhausted()) {
    if (cursor.hasNext()) {
        printjson(cursor.next());
    }
}

var collection = new MongoClient("<connection-string>")
    .GetDatabase("<database-name>")
    .GetCollection<BsonDocument>("<collection-name>");

var pipeline = new EmptyPipelineDefinition<ChangeStreamDocument<BsonDocument>>()
    .Match(change => 
        change.OperationType == ChangeStreamOperationType.Insert || 
        change.OperationType == ChangeStreamOperationType.Update || 
        change.OperationType == ChangeStreamOperationType.Replace
    )
    .AppendStage<ChangeStreamDocument<BsonDocument>, ChangeStreamDocument<BsonDocument>, BsonDocument>(
        @"{ 
            $project: { 
                '_id': 1, 
                'fullDocument': 1, 
                'ns': 1, 
                'documentKey': 1 
            }
        }"
    );

ChangeStreamOptions options = new ()
{
    FullDocument = ChangeStreamFullDocumentOption.UpdateLookup
};

using IChangeStreamCursor<BsonDocument> enumerator = collection.Watch(
    pipeline, 
    options
);

Console.WriteLine("Waiting for changes...");
while (enumerator.MoveNext())
{
    IEnumerable<BsonDocument> changes = enumerator.Current;
    foreach(BsonDocument change in changes)
    {
        Console.WriteLine(change);
    }  
}

Bson match = Aggregates.match(Filters.in("operationType", asList("update", "replace", "insert")));

// Pick the field you are most interested in
Bson project = Aggregates.project(fields(include("_id", "ns", "documentKey", "fullDocument")));

// This variable is for second example
BsonDocument resumeToken = null;

// Now time to build the pipeline
List<Bson> pipeline = Arrays.asList(match, project);

//#1 Simple example to seek changes

// Create cursor with update_lookup
MongoChangeStreamCursor<ChangeStreamDocument<org.bson.Document>> cursor = collection.watch(pipeline)
        .fullDocument(FullDocument.UPDATE_LOOKUP).cursor();

Document document = new Document("name", "doc-in-step-1-" + Math.random());
collection.insertOne(document);

while (cursor.hasNext()) {
    // There you go, we got the change document.
    ChangeStreamDocument<Document> csDoc = cursor.next();

    // Let is pick the token which will help us resuming
    // You can save this token in any persistent storage and retrieve it later
    resumeToken = csDoc.getResumeToken();
    //Printing the token
    System.out.println(resumeToken);
    
    //Printing the document.
    System.out.println(csDoc.getFullDocument());
    //This break is intentional but in real project feel free to remove it.
    break;
}

cursor.close();

Módosítások egyetlen szegmensen belül

Az alábbi példa bemutatja, hogyan kérhet le módosításokat az egyes szegmensek elemein. Ez a példa beolvasja azokat az elemeket, amelyeknél a szegmenskulcs értéke "a" és a szegmenskulcs értéke "1". Lehetséges, hogy a különböző ügyfelek párhuzamosan olvassák a különböző szegmensek módosításait.

var cursor = db.coll.watch(
    [
        { 
            $match: { 
                $and: [
                    { "fullDocument.a": 1 }, 
                    { "operationType": { $in: ["insert", "update", "replace"] } }
                ]
            }
        },
        { $project: { "_id": 1, "fullDocument": 1, "ns": 1, "documentKey": 1} }
    ],
    { fullDocument: "updateLookup" });

Streamek módosítása skálázása

A változásstreamek nagy léptékű használata esetén a legjobb, ha egyenletesen elosztja a terhelést. Használja a GetChangeStreamTokens egyéni parancsot a terhelés fizikai szegmensek/partíciók közötti elosztásához.

Jelenlegi korlátozások

A változásstreamek használatakor a következő korlátozások érvényesek:

• A operationType kimeneti dokumentum még nem támogatja a tulajdonságokat és updateDescription a tulajdonságokat.
• A insert, updateés replace a művelettípusok jelenleg támogatottak. A törlési művelet vagy más események azonban még nem támogatottak.

Ezen korlátozások miatt a $match szakasz, a $project szakasz és a fullDocument beállítások szükségesek az előző példákban látható módon.

Az Azure Cosmos DB NOSQL-hez készült API-jának változáscsatornájával ellentétben nincs külön változáscsatorna-feldolgozó kódtár a változásstreamek használatához, vagy nincs szükség bérlettárolóra. Az Azure Functions-eseményindítók jelenleg nem támogatják a változásstreamek feldolgozását.

Hibakezelés

A változásstreamek használatakor a következő hibakódok és üzenetek támogatottak:

• HTTP-hibakód: 16500 – A változásstream szabályozása után üres lapot ad vissza.

• NamespaceNotFound (OperationType Invalidate) – Ha nem létező gyűjteményen futtatja a változásstreamet, vagy ha a gyűjtemény elvetve van, a NamespaceNotFound rendszer hibát ad vissza. Mivel a operationType tulajdonság nem adható vissza a kimeneti dokumentumban, a operationType Invalidate hiba helyett a rendszer a NamespaceNotFound hibát adja vissza.

Következő lépések

• Élettartam az adatok automatikus lejáratához az Azure Cosmos DB MongoDB-hez készült API-jában
• Indexelés az Azure Cosmos DB MongoDB-hez készült API-jában