MongoDB 3.6: Change Stream

V minulém roce jsem se pokoušel replikovat změny v datach pod MSSQL. Šlo mi o to něajak v reálném čase odchytávat SQL příkazy INSERT, UPDATE a DELETE, abych mohl na úrvni změněných řádků v tabulce provedené změny exportovat ven z MSSQL databáze. Našel jsem nějakou nativní implementaci Microsoftu přímo v aplikaci databázového server… Funkcionalita dostupná až v poslední verzi serveru… Práce s ní strašný opruz s nutnou dávkou dalšího úsilí jak se k datům dostat. A o tom, že to budou data v skoro reálném čase si nechte zdát…

Change Stream

O to větší radost jsem měl radost z jedné z mnoho novinek v MongoDB verze 3.6, kterou je Change Stream, což je přesně to, co jsem chtěl po MSSQL. Elegance s kterou to řeší MongoDB je úžasná. Celá logika a funkcionalita je postavena nad oplogem (kde jinde…) a chová se přesně jak byste čekali: oplog je zpracováván na úrovni replica setu pro migraci informací mezi jednotlivými servery od primárního serveru k sekundárům. Ovladač MongoDB má nově funkci watch ,nad objektem collection, která vrací data z oplogu. Jak jednoduché.

Apka pro vkládáni dat do MongoDB databáze

Touto jednoduchou aplikací jen vkládám jednotlivé dokumenty do sledované kolekce. Stejně tak bych to mohl dělat přímo přes Mongo shell, nebo v reálu se mi o toto bude starat samotná aplikace, která daná data přímo spravuje.

 {
    if (err) console.error(err);
    else {
        let db = connecion.db(dbName);
        let coll = db.collection(collectionName);
        coll.insert(uzivatel, (err, res) => {
            if (err) console.error(err);
            else {
                console.log('vlozena nova firma');
            }
            connecion.close(err => {
                if (err) console.error('nepodarilo se odpojit od monga...');
            });
        });
    }
});

Nic složitého: připojíte se MongoDB replica setu a vložíte do dané collection jeden záznam.

Apka pro sledování změněných dat

'use strict';

const mongodb = require('mongodb');

const mongoUrl = 'mongodb://localhost:27017,localhost:27018,localhost:27019?replicaSet=rs';
const dbName = 'test';
const collectionName = 'uzivatel';

mongodb.MongoClient.connect(mongoUrl, (err, connecion) => {
    if (err) console.error(err);
    else {
        let db = connecion.db(dbName);
        let coll = db.collection(collectionName);
        const changeStream = coll.watch({ full_document: 'updateLookup' });

        changeStream.on('change', change => {
            console.log(change);
        });

    }
});

Apka se opět připojuje k MongoDB a ve funcki watch nad sledovanou collection dostává dat o provedených změnách. Jednodušeji to snad už nejde…

Kód jsem napsal v JavaScriptu pro NodeJS. Přepsat jej do Pythonu je brnkačka a výsledný kód, vzhledem k absenci obslužných callbackových funkcí bude mnohem kratší. Nicméně příjdete o asynchronní zpracování, které vám nabízí NodeJS.