Replikering på højt niveau i Tarantool DBMS

Hej, jeg laver applikationer til DBMS Tarantværktøj er en platform udviklet af Mail.ru Group, der kombinerer en højtydende DBMS og en applikationsserver på Lua-sproget. Den høje hastighed af løsninger baseret på Tarantool opnås, især på grund af understøttelse af in-memory mode af DBMS og evnen til at udføre applikationsforretningslogik i et enkelt adresserum med data. Samtidig sikres datapersistens ved hjælp af ACID-transaktioner (en WAL-log vedligeholdes på disken). Tarantool har indbygget understøttelse af replikering og sønderdeling. Fra version 2.1 understøttes forespørgsler i SQL-sprog. Tarantool er open source og licenseret under Simplified BSD-licensen. Der er også en kommerciel Enterprise-version.

Mærk kraften! (…aka nyd forestillingen)

Alt ovenstående gør Tarantool til en attraktiv platform til at skabe højbelastede applikationer, der fungerer med databaser. I sådanne applikationer er der ofte behov for datareplikering.

Som nævnt ovenfor har Tarantool indbygget datareplikering. Princippet for dets funktion er at sekventielt udføre på replikaer alle transaktioner indeholdt i masterloggen (WAL). Normalt en sådan replikation (vi vil yderligere kalde det lavt niveau) bruges til at sikre applikationsfejltolerance og/eller til at fordele læsebelastningen mellem klynge noder.

Ris. 1. Replikering i en klynge

Et eksempel på et alternativt scenarie ville være overførsel af data oprettet i en database til en anden database til behandling/overvågning. I sidstnævnte tilfælde kan en mere bekvem løsning være at bruge højt niveau replikering - datareplikering på applikationens forretningslogikniveau. De der. Vi bruger ikke en færdiglavet løsning indbygget i DBMS, men implementerer replikering på egen hånd i den applikation, vi er ved at udvikle. Denne tilgang har både fordele og ulemper. Lad os liste fordelene.

1. Trafikbesparelser:

• Du kan ikke overføre alle data, men kun en del af dem (f.eks. kan du kun overføre nogle tabeller, nogle af deres kolonner eller poster, der opfylder et bestemt kriterium);
• I modsætning til replikering på lavt niveau, som udføres kontinuerligt i asynkron (implementeret i den nuværende version af Tarantool - 1.10) eller synkron (skal implementeres i efterfølgende versioner af Tarantool), kan replikering på højt niveau udføres i sessioner (dvs. applikationen synkroniserer først dataene - en udvekslingssessionsdata, derefter er der en pause i replikeringen, hvorefter den næste udvekslingssession finder sted, osv.);
• hvis en post har ændret sig flere gange, kan du kun overføre dens seneste version (i modsætning til replikering på lavt niveau, hvor alle ændringer foretaget på masteren vil blive afspillet sekventielt på replikaerne).

2. Der er ingen problemer med at implementere HTTP-udveksling, som giver dig mulighed for at synkronisere fjerndatabaser.

Ris. 2. Replikering over HTTP

3. Databasestrukturerne, som data overføres imellem, behøver ikke at være ens (desuden er det i det generelle tilfælde endda muligt at bruge forskellige DBMS'er, programmeringssprog, platforme osv.).

Ris. 3. Replikation i heterogene systemer

Ulempen er, at programmering i gennemsnit er sværere/dyrere end konfiguration, og i stedet for at tilpasse den indbyggede funktionalitet, bliver du nødt til at implementere din egen.

Hvis ovenstående fordele i din situation er afgørende (eller er en nødvendig betingelse), så giver det mening at bruge replikering på højt niveau. Lad os se på flere måder at implementere datareplikering på højt niveau i Tarantool DBMS.

Trafikminimering

Så en af ​​fordelene ved replikering på højt niveau er trafikbesparelser. For at denne fordel kan realiseres fuldt ud, er det nødvendigt at minimere mængden af ​​data, der overføres under hver udvekslingssession. Selvfølgelig skal vi ikke glemme, at i slutningen af ​​sessionen skal datamodtageren synkroniseres med kilden (i det mindste for den del af dataene, der er involveret i replikering).

Hvordan minimerer man mængden af ​​data, der overføres under replikering på højt niveau? En ligetil løsning kunne være at vælge data efter dato og klokkeslæt. For at gøre dette kan du bruge det dato-tidsfelt, der allerede findes i tabellen (hvis det findes). For eksempel kan et "ordre"-dokument have et felt "påkrævet ordreudførelsestid" - delivery_time. Problemet med denne løsning er, at værdierne i dette felt ikke behøver at være i den rækkefølge, der svarer til oprettelsen af ​​ordrer. Så vi kan ikke huske den maksimale feltværdi delivery_time, transmitteret under den forrige udvekslingssession, og under den næste udvekslingssession vælges alle poster med en højere feltværdi delivery_time. Poster med en lavere feltværdi kan være blevet tilføjet mellem udvekslingssessioner delivery_time. Også ordren kunne have undergået ændringer, som dog ikke påvirkede feltet delivery_time. I begge tilfælde vil ændringerne ikke blive overført fra kilden til destinationen. For at løse disse problemer bliver vi nødt til at overføre data "overlappende". De der. i hver udvekslingssession overfører vi alle data med feltværdien delivery_time, der overskrider et punkt i fortiden (f.eks. N timer fra det aktuelle øjeblik). Det er dog indlysende, at for store systemer er denne tilgang yderst overflødig og kan reducere de trafikbesparelser, vi stræber efter, til ingenting. Derudover har tabellen, der overføres, muligvis ikke et felt tilknyttet en dato-tid.

En anden løsning, mere kompleks med hensyn til implementering, er at bekræfte modtagelsen af ​​data. I dette tilfælde, under hver udvekslingssession, transmitteres alle data, hvis modtagelse ikke er bekræftet af modtageren. For at implementere dette skal du tilføje en boolsk kolonne til kildetabellen (f.eks. is_transferred). Hvis modtageren bekræfter modtagelsen af ​​posten, tager det tilsvarende felt værdien true, hvorefter indgangen ikke længere er involveret i udvekslinger. Denne implementeringsmulighed har følgende ulemper. Først skal der for hver overført post genereres og sendes en bekræftelse. Det kan groft sagt sammenlignes med at fordoble mængden af ​​overført data og føre til en fordobling af antallet af rundturer. For det andet er der ingen mulighed for at sende den samme post til flere modtagere (den første modtager vil bekræfte modtagelsen for sig selv og for alle de andre).

En metode, der ikke har de ovennævnte ulemper, er at tilføje en kolonne til den overførte tabel for at spore ændringer i dens rækker. En sådan kolonne kan være af dato-tid-typen og skal indstilles/opdateres af applikationen til det aktuelle tidspunkt, hver gang poster tilføjes/ændres (atomisk med tilføjelsen/ændringen). Lad os som et eksempel kalde kolonnen update_time. Ved at gemme den maksimale feltværdi for denne kolonne for de overførte poster, kan vi starte den næste udvekslingssession med denne værdi (vælg poster med feltværdien update_time, der overstiger den tidligere gemte værdi). Problemet med sidstnævnte tilgang er, at dataændringer kan forekomme i batches. Som et resultat af feltværdierne i kolonnen update_time er muligvis ikke unik. Denne kolonne kan således ikke bruges til portioneret (side-for-side) dataoutput. For at vise data side for side, skal du opfinde yderligere mekanismer, der højst sandsynligt vil have meget lav effektivitet (f.eks. hente fra databasen alle poster med værdien update_time højere end en given og producerer et vist antal poster, startende fra en vis offset fra begyndelsen af ​​prøven).

Du kan forbedre effektiviteten af ​​dataoverførsel ved at forbedre den tidligere tilgang en smule. For at gøre dette vil vi bruge heltalstypen (langt heltal) som kolonnefeltværdierne til sporing af ændringer. Lad os navngive kolonnen row_ver. Feltværdien for denne kolonne skal stadig indstilles/opdateres hver gang en post oprettes/ændres. Men i dette tilfælde vil feltet ikke blive tildelt den aktuelle dato-tid, men værdien af ​​en eller anden tæller, forhøjet med én. Som et resultat, kolonnen row_ver vil indeholde unikke værdier og kan ikke kun bruges til at vise "delta"-data (data tilføjet/ændret siden slutningen af ​​den forrige udvekslingssession), men også til enkelt og effektivt at opdele dem i sider.

Den sidste foreslåede metode til at minimere mængden af ​​data, der overføres inden for rammerne af replikering på højt niveau, forekommer mig at være den mest optimale og universelle. Lad os se på det mere detaljeret.

Videregivelse af data ved hjælp af en rækkeversionstæller

Implementering af server/master delen

I MS SQL Server er der en speciel kolonnetype til at implementere denne tilgang - rowversion. Hver database har en tæller, der stiger med én, hver gang en post tilføjes/ændres i en tabel, der har en kolonne som rowversion. Værdien af ​​denne tæller tildeles automatisk til feltet i denne kolonne i den tilføjede/ændrede post. Tarantool DBMS har ikke en lignende indbygget mekanisme. Men i Tarantool er det ikke svært at implementere det manuelt. Lad os se på, hvordan dette gøres.

Først lidt terminologi: Tabeller i Tarantool kaldes mellemrum, og poster kaldes tuples. I Tarantool kan du oprette sekvenser. Sekvenser er intet andet end navngivne generatorer af ordnede heltalsværdier. De der. det er præcis, hvad vi har brug for til vores formål. Nedenfor vil vi oprette en sådan sekvens.

Før du udfører nogen databaseoperation i Tarantool, skal du køre følgende kommando:

box.cfg{}

Som et resultat vil Tarantool begynde at skrive database-øjebliksbilleder og transaktionslogfiler til den aktuelle mappe.

Lad os skabe en sekvens row_version:

box.schema.sequence.create('row_version',
    { if_not_exists = true })

valgmulighed if_not_exists tillader oprettelsesscriptet at blive udført flere gange: hvis objektet eksisterer, vil Tarantool ikke forsøge at oprette det igen. Denne mulighed vil blive brugt i alle efterfølgende DDL-kommandoer.

Lad os skabe et rum som et eksempel.

box.schema.space.create('goods', {
    format = {
        {
            name = 'id',
            type = 'unsigned'

        },
        {
            name = 'name',
            type = 'string'

        },
        {
            name = 'code',
            type = 'unsigned'

        },
        {
            name = 'row_ver',
            type = 'unsigned'

        }
    },
    if_not_exists = true
})

Her sætter vi navnet på rummet (goods), feltnavne og deres typer.

Auto-inkrementerende felter i Tarantool oprettes også ved hjælp af sekvenser. Lad os oprette en primær nøgle med automatisk stigning efter felt id:

box.schema.sequence.create('goods_id',
    { if_not_exists = true })
box.space.goods:create_index('primary', {
    parts = { 'id' },
    sequence = 'goods_id',
    unique = true,
    type = 'HASH',
    if_not_exists = true
})

Tarantool understøtter flere typer indekser. De mest brugte indekser er TREE- og HASH-typer, som er baseret på strukturer svarende til navnet. TREE er den mest alsidige indekstype. Det giver dig mulighed for at hente data på en organiseret måde. Men til udvælgelse af ligestilling er HASH mere egnet. Derfor er det tilrådeligt at bruge HASH til den primære nøgle (hvilket er, hvad vi gjorde).

For at bruge kolonnen row_ver for at overføre ændrede data skal du binde sekvensværdier til felterne i denne kolonne row_ver. Men i modsætning til den primære nøgle, kolonnens feltværdi row_ver bør stige med én, ikke kun ved tilføjelse af nye poster, men også ved ændring af eksisterende. Du kan bruge triggere til dette. Tarantool har to typer rumudløsere: before_replace и on_replace. Triggere udløses, hver gang dataene i rummet ændres (for hver tuple, der påvirkes af ændringerne, startes en triggerfunktion). I modsætning til on_replace, before_replace-triggere giver dig mulighed for at ændre dataene for den tuple, som triggeren udføres for. Derfor passer den sidste type triggere os.

box.space.goods:before_replace(function(old, new)
    return box.tuple.new({new[1], new[2], new[3],
        box.sequence.row_version:next()})
end)

Følgende trigger erstatter feltværdien row_ver gemt tuple til næste værdi af sekvensen row_version.

For at kunne udtrække data fra rummet goods efter kolonne row_ver, lad os oprette et indeks:

box.space.goods:create_index('row_ver', {
    parts = { 'row_ver' },
    unique = true,
    type = 'TREE',
    if_not_exists = true
})

Indekstype - træ (TREE), fordi vi bliver nødt til at udtrække dataene i stigende rækkefølge af værdierne i kolonnen row_ver.

Lad os tilføje nogle data til rummet:

box.space.goods:insert{nil, 'pen', 123}
box.space.goods:insert{nil, 'pencil', 321}
box.space.goods:insert{nil, 'brush', 100}
box.space.goods:insert{nil, 'watercolour', 456}
box.space.goods:insert{nil, 'album', 101}
box.space.goods:insert{nil, 'notebook', 800}
box.space.goods:insert{nil, 'rubber', 531}
box.space.goods:insert{nil, 'ruler', 135}

Fordi Det første felt er en tæller med automatisk inkrementering; vi sender i stedet nul. Tarantool vil automatisk erstatte den næste værdi. På samme måde som værdien af ​​kolonnefelterne row_ver du kan passere nul - eller slet ikke angive værdien, fordi denne kolonne indtager den sidste position i rummet.

Lad os tjekke indsættelsesresultatet:

tarantool> box.space.goods:select()
---
- - [1, 'pen', 123, 1]
  - [2, 'pencil', 321, 2]
  - [3, 'brush', 100, 3]
  - [4, 'watercolour', 456, 4]
  - [5, 'album', 101, 5]
  - [6, 'notebook', 800, 6]
  - [7, 'rubber', 531, 7]
  - [8, 'ruler', 135, 8]
...

Som du kan se, udfyldes første og sidste felt automatisk. Nu bliver det nemt at skrive en funktion til side-for-side upload af pladsændringer goods:

local page_size = 5
local function get_goods(row_ver)
    local index = box.space.goods.index.row_ver
    local goods = {}
    local counter = 0
    for _, tuple in index:pairs(row_ver, {
        iterator = 'GT' }) do
        local obj = tuple:tomap({ names_only = true })
        table.insert(goods, obj)
        counter = counter + 1
        if counter >= page_size then
            break
        end
    end
    return goods
end

Funktionen tager værdien som parameter row_ver, hvorfra det er nødvendigt at fjerne ændringer, og returnerer en del af de ændrede data.

Datasampling i Tarantool udføres gennem indekser. Fungere get_goods bruger en iterator efter indeks row_ver at modtage ændrede data. Iteratortypen er GT (Større end, større end). Dette betyder, at iteratoren sekventielt vil krydse indeksværdierne fra den beståede nøgle (feltværdi row_ver).

Iteratoren returnerer tupler. For efterfølgende at kunne overføre data via HTTP, er det nødvendigt at konvertere tuplerne til en struktur, der er praktisk til efterfølgende serialisering. Eksemplet bruger standardfunktionen til dette tomap. I stedet for at bruge tomap du kan skrive din egen funktion. For eksempel vil vi måske omdøbe et felt name, ikke passere feltet code og tilføje et felt comment:

local function unflatten_goods(tuple)
    local obj = {}
    obj.id = tuple.id
    obj.goods_name = tuple.name
    obj.comment = 'some comment'
    obj.row_ver = tuple.row_ver
    return obj
end

Sidestørrelsen af ​​outputdataene (antallet af poster i én del) bestemmes af variablen page_size. I eksemplet værdien page_size er 5. I et rigtigt program betyder sidestørrelsen normalt mere. Det afhænger af den gennemsnitlige størrelse af space tuple. Den optimale sidestørrelse kan bestemmes empirisk ved at måle dataoverførselstid. Jo større sidestørrelsen er, jo mindre er antallet af rundrejser mellem afsender- og modtagesiden. På denne måde kan du reducere den samlede tid til at downloade ændringer. Men hvis sidestørrelsen er for stor, vil vi bruge for lang tid på serveren med at serialisere prøven. Som følge heraf kan der være forsinkelser i behandlingen af ​​andre anmodninger, der kommer til serveren. Parameter page_size kan indlæses fra konfigurationsfilen. For hver overført plads kan du indstille sin egen værdi. For de fleste mellemrum kan standardværdien (f.eks. 100) dog være passende.

Lad os udføre funktionen get_goods:

tarantool> get_goods(0)

---
- - row_ver: 1
    code: 123
    name: pen
    id: 1
  - row_ver: 2
    code: 321
    name: pencil
    id: 2
  - row_ver: 3
    code: 100
    name: brush
    id: 3
  - row_ver: 4
    code: 456
    name: watercolour
    id: 4
  - row_ver: 5
    code: 101
    name: album
    id: 5
...

Lad os tage feltværdien row_ver fra sidste linje og kald funktionen igen:

tarantool> get_goods(5)

---
- - row_ver: 6
    code: 800
    name: notebook
    id: 6
  - row_ver: 7
    code: 531
    name: rubber
    id: 7
  - row_ver: 8
    code: 135
    name: ruler
    id: 8
...

Endnu engang:

tarantool> get_goods(8)
---
- []
...

Som du kan se, returnerer funktionen alle pladsposter side for side, når den bruges på denne måde goods. Den sidste side efterfølges af et tomt valg.

Lad os foretage ændringer i rummet:

box.space.goods:update(4, {{'=', 6, 'copybook'}})
box.space.goods:insert{nil, 'clip', 234}
box.space.goods:insert{nil, 'folder', 432}

Vi har ændret feltværdien name for én post og tilføjet to nye poster.

Lad os gentage det sidste funktionskald:

tarantool> get_goods(8)
---



- - row_ver: 9
    code: 800
    name: copybook
    id: 6
  - row_ver: 10
    code: 234
    name: clip
    id: 9
  - row_ver: 11
    code: 432
    name: folder
    id: 10
...

Funktionen returnerede de ændrede og tilføjede poster. Altså funktionen get_goods giver dig mulighed for at modtage data, der er ændret siden det sidste opkald, hvilket er grundlaget for den replikeringsmetode, der overvejes.

Vi vil lade udstedelsen af ​​resultater via HTTP i form af JSON være uden for denne artikels omfang. Det kan du læse om her: https://habr.com/ru/company/mailru/blog/272141/

Implementering af klient/slave delen

Lad os se på, hvordan den modtagende sides implementering ser ud. Lad os oprette en plads på modtagersiden til at gemme de downloadede data:

box.schema.space.create('goods', {
    format = {
        {
            name = 'id',
            type = 'unsigned'

        },
        {
            name = 'name',
            type = 'string'

        },
        {
            name = 'code',
            type = 'unsigned'

        }
    },
    if_not_exists = true
})

box.space.goods:create_index('primary', {
    parts = { 'id' },
    sequence = 'goods_id',
    unique = true,
    type = 'HASH',
    if_not_exists = true
})

Rummets struktur ligner strukturen af ​​rummet i kilden. Men da vi ikke kommer til at videregive de modtagne data andre steder, er kolonnen row_ver er ikke i modtagerens plads. I marken id kildeidentifikatorer vil blive registreret. På modtagersiden er der derfor ingen grund til at gøre den automatisk inkrementerende.

Derudover har vi brug for et rum til at gemme værdier row_ver:

box.schema.space.create('row_ver', {
    format = {
        {
            name = 'space_name',
            type = 'string'

        },
        {
            name = 'value',
            type = 'string'

        }
    },
    if_not_exists = true
})

box.space.row_ver:create_index('primary', {
    parts = { 'space_name' },
    unique = true,
    type = 'HASH',
    if_not_exists = true
})

For hver indlæst plads (felt space_name) gemmer vi den sidst indlæste værdi her row_ver (Mark value). Kolonnen fungerer som den primære nøgle space_name.

Lad os oprette en funktion til at indlæse pladsdata goods via HTTP. For at gøre dette har vi brug for et bibliotek, der implementerer en HTTP-klient. Følgende linje indlæser biblioteket og instansierer HTTP-klienten:

local http_client = require('http.client').new()

Vi har også brug for et bibliotek til json-deserialisering:

local json = require('json')

Dette er nok til at oprette en dataindlæsningsfunktion:

local function load_data(url, row_ver)
    local url = ('%s?rowVer=%s'):format(url,
        tostring(row_ver))
    local body = nil
    local data = http_client:request('GET', url, body, {
        keepalive_idle =  1,
        keepalive_interval = 1
    })
    return json.decode(data.body)
end

Funktionen udfører en HTTP-anmodning til url-adressen og sender den row_ver som en parameter og returnerer det deserialiserede resultat af anmodningen.

Funktionen til at gemme modtagne data ser således ud:

local function save_goods(goods)
    local n = #goods
    box.atomic(function()
        for i = 1, n do
            local obj = goods[i]
            box.space.goods:put(
                obj.id, obj.name, obj.code)
        end
    end)
end

Cyklus for at gemme data i rummet goods placeres i en transaktion (funktionen bruges til dette box.atomic) for at reducere antallet af diskoperationer.

Endelig den lokale rumsynkroniseringsfunktion goods med en kilde kan du implementere det sådan her:

local function sync_goods()
    local tuple = box.space.row_ver:get('goods')
    local row_ver = tuple and tuple.value or 0

    —— set your url here:
    local url = 'http://127.0.0.1:81/test/goods/list'

    while true do
        local goods = load_goods(url, row_ver)

        local count = #goods
        if count == 0 then
            return
        end

        save_goods(goods)

        row_ver = goods[count].rowVer
        box.space.row_ver:put({'goods', row_ver})
    end
end

Først læser vi den tidligere gemte værdi row_ver for plads goods. Hvis det mangler (den første udvekslingssession), så tager vi det som row_ver nul. Næste i cyklussen udfører vi en side-for-side download af de ændrede data fra kilden på den angivne url. Ved hver iteration gemmer vi de modtagne data på det relevante lokale område og opdaterer værdien row_ver (i rummet row_ver og i variablen row_ver) - tag værdien row_ver fra den sidste linje med indlæste data.

For at beskytte mod utilsigtet looping (i tilfælde af en fejl i programmet), skal loopen while kan erstattes af for:

for _ = 1, max_req do ...

Som et resultat af at udføre funktionen sync_goods plads goods modtageren vil indeholde de seneste versioner af alle rumposter goods i kilden.

Datasletning kan naturligvis ikke udsendes på denne måde. Hvis et sådant behov eksisterer, kan du bruge et slettemærke. Tilføj plads goods boolesk felt is_deleted og i stedet for fysisk at slette en post, bruger vi logisk sletning - vi sætter feltværdien is_deleted i betydning true. Nogle gange i stedet for et boolsk felt is_deleted det er mere bekvemt at bruge feltet deleted, som gemmer dato-klokkeslæt for den logiske sletning af posten. Efter at have udført en logisk sletning vil posten, der er markeret til sletning, blive overført fra kilden til destinationen (i henhold til logikken beskrevet ovenfor).

sekvens row_ver kan bruges til at overføre data fra andre rum: der er ikke behov for at oprette en separat sekvens for hvert overført rum.

Vi kiggede på en effektiv måde at replikere data på højt niveau i applikationer, der bruger Tarantool DBMS.

Fund

1. Tarantool DBMS er et attraktivt, lovende produkt til at skabe højbelastningsapplikationer.
2. Datareplikering på højt niveau har en række fordele i forhold til replikering på lavt niveau.
3. Den højniveau-replikeringsmetode, der diskuteres i artiklen, giver dig mulighed for at minimere mængden af ​​overførte data ved kun at overføre de poster, der er ændret siden den sidste udvekslingssession.

Kilde: www.habr.com