Replikering på høyt nivå i Tarantool DBMS

Hei, jeg lager applikasjoner for DBMS Tarantool er en plattform utviklet av Mail.ru Group som kombinerer en høyytelses DBMS og en applikasjonsserver på Lua-språket. Den høye hastigheten til løsninger basert på Tarantool oppnås, spesielt på grunn av støtte for minnemodusen til DBMS og muligheten til å utføre applikasjonsforretningslogikk i et enkelt adresserom med data. Samtidig sikres datapersistens ved hjelp av ACID-transaksjoner (en WAL-logg opprettholdes på disken). Tarantool har innebygd støtte for replikering og skjæring. Fra og med versjon 2.1 støttes spørringer i SQL-språk. Tarantool er åpen kildekode og lisensiert under Simplified BSD-lisensen. Det finnes også en kommersiell Enterprise-versjon.

Føl kraften! (…aka nyt forestillingen)

Alt det ovennevnte gjør Tarantool til en attraktiv plattform for å lage høybelastningsapplikasjoner som fungerer med databaser. I slike applikasjoner er det ofte behov for datareplikering.

Som nevnt ovenfor har Tarantool innebygd datareplikering. Prinsippet for operasjonen er å sekvensielt utføre alle transaksjoner i hovedloggen (WAL) på replikaer. Vanligvis slik replikering (vi vil videre kalle det lavt nivå) brukes til å sikre applikasjonsfeiltoleranse og/eller for å fordele lesebelastningen mellom klyngenoder.

Ris. 1. Replikering i en klynge

Et eksempel på et alternativt scenario vil være å overføre data opprettet i en database til en annen database for behandling/overvåking. I sistnevnte tilfelle kan en mer praktisk løsning være å bruke høy level replikering - datareplikering på applikasjons forretningslogikknivå. De. Vi bruker ikke en ferdig løsning innebygd i DBMS, men implementerer replikering på egen hånd innenfor applikasjonen vi utvikler. Denne tilnærmingen har både fordeler og ulemper. La oss liste opp fordelene.

1. Trafikkbesparelser:

• Du kan ikke overføre alle dataene, men bare deler av dem (for eksempel kan du overføre bare noen tabeller, noen av deres kolonner eller poster som oppfyller et bestemt kriterium);
• I motsetning til replikering på lavt nivå, som utføres kontinuerlig i asynkron (implementert i gjeldende versjon av Tarantool - 1.10) eller synkron (som skal implementeres i påfølgende versjoner av Tarantool), kan replikering på høyt nivå utføres i økter (dvs. applikasjonen synkroniserer først dataene - en utvekslingsøktdata, deretter er det en pause i replikering, hvoretter neste utvekslingsøkt skjer, etc.);
• hvis en post har endret seg flere ganger, kan du overføre bare den nyeste versjonen (i motsetning til replikering på lavt nivå, der alle endringer som er gjort på masteren vil bli spilt av sekvensielt på replikaene).

2. Det er ingen problemer med å implementere HTTP-utveksling, som lar deg synkronisere eksterne databaser.

Ris. 2. Replikering over HTTP

3. Databasestrukturene som data overføres mellom trenger ikke å være de samme (i det generelle tilfellet er det dessuten mulig å bruke forskjellige DBMS-er, programmeringsspråk, plattformer osv.).

Ris. 3. Replikasjon i heterogene systemer

Ulempen er at programmering i gjennomsnitt er vanskeligere/kostbart enn konfigurasjon, og i stedet for å tilpasse den innebygde funksjonaliteten, må du implementere din egen.

Hvis fordelene ovenfor er avgjørende i din situasjon (eller er en nødvendig betingelse), er det fornuftig å bruke replikering på høyt nivå. La oss se på flere måter å implementere datareplikering på høyt nivå i Tarantool DBMS.

Trafikkminimering

Så en av fordelene med replikering på høyt nivå er trafikkbesparelser. For at denne fordelen skal realiseres fullt ut, er det nødvendig å minimere mengden data som overføres under hver utvekslingsøkt. Selvfølgelig bør vi ikke glemme at på slutten av økten må datamottakeren synkroniseres med kilden (i det minste for den delen av dataene som er involvert i replikering).

Hvordan minimere mengden data som overføres under replikering på høyt nivå? En grei løsning kan være å velge data etter dato og klokkeslett. For å gjøre dette kan du bruke dato-klokkeslett-feltet som allerede finnes i tabellen (hvis det finnes). For eksempel kan et "ordre"-dokument ha et felt "påkrevd ordreutførelsestid" - delivery_time. Problemet med denne løsningen er at verdiene i dette feltet ikke trenger å være i sekvensen som tilsvarer opprettelsen av bestillinger. Så vi kan ikke huske den maksimale feltverdien delivery_time, overført under forrige utvekslingsøkt, og under neste utvekslingsøkt velg alle poster med en høyere feltverdi delivery_time. Poster med lavere feltverdi kan ha blitt lagt til mellom utvekslingsøktene delivery_time. Dessuten kunne ordren ha gjennomgått endringer, som likevel ikke påvirket feltet delivery_time. I begge tilfeller vil ikke endringene bli overført fra kilden til destinasjonen. For å løse disse problemene må vi overføre data "overlappende". De. i hver utvekslingsøkt vil vi overføre alle data med feltverdien delivery_time, som overskrider et punkt i fortiden (for eksempel N timer fra nåværende øyeblikk). Det er imidlertid åpenbart at for store systemer er denne tilnærmingen svært overflødig og kan redusere trafikkbesparelsene som vi streber etter til ingenting. I tillegg kan det hende at tabellen som overføres ikke har et felt knyttet til en dato-klokkeslett.

En annen løsning, mer kompleks med tanke på implementering, er å bekrefte mottak av data. I dette tilfellet, under hver utvekslingsøkt, overføres alle data, mottakeren av disse er ikke bekreftet av mottakeren. For å implementere dette må du legge til en boolsk kolonne i kildetabellen (f.eks. is_transferred). Hvis mottakeren bekrefter mottak av posten, tar det tilsvarende feltet verdien true, hvoretter oppføringen ikke lenger er involvert i utveksling. Dette implementeringsalternativet har følgende ulemper. Først, for hver post som overføres, må det genereres og sendes en bekreftelse. Grovt sett kan dette sammenlignes med å doble mengden data som overføres og føre til en dobling av antall rundturer. For det andre er det ingen mulighet for å sende samme post til flere mottakere (den første mottakeren som mottar vil bekrefte mottak for seg selv og for alle de andre).

En metode som ikke har de ulempene som er gitt ovenfor, er å legge til en kolonne i den overførte tabellen for å spore endringer i dens rader. En slik kolonne kan være av dato-tid-type og må settes/oppdateres av applikasjonen til gjeldende klokkeslett hver gang poster legges til/endres (atomisk med tillegget/endringen). Som et eksempel, la oss kalle kolonnen update_time. Ved å lagre den maksimale feltverdien for denne kolonnen for de overførte postene, kan vi starte neste utvekslingsøkt med denne verdien (velg poster med feltverdien update_time, som overskrider den tidligere lagrede verdien). Problemet med sistnevnte tilnærming er at dataendringer kan forekomme i batcher. Som et resultat av feltverdiene i kolonnen update_time er kanskje ikke unikt. Dermed kan ikke denne kolonnen brukes for porsjonert (side-for-side) datautgang. For å vise data side for side, må du finne opp flere mekanismer som mest sannsynlig vil ha svært lav effektivitet (for eksempel å hente fra databasen alle poster med verdien update_time høyere enn en gitt og produserer et visst antall poster, med utgangspunkt i en viss offset fra begynnelsen av prøven).

Du kan forbedre effektiviteten til dataoverføring ved å forbedre den forrige tilnærmingen litt. For å gjøre dette vil vi bruke heltallstypen (langt heltall) som kolonnefeltverdier for sporing av endringer. La oss gi kolonnen et navn row_ver. Feltverdien til denne kolonnen må fortsatt settes/oppdateres hver gang en post opprettes/endres. Men i dette tilfellet vil ikke feltet bli tildelt gjeldende dato-klokkeslett, men verdien til en teller, økt med én. Som et resultat, kolonnen row_ver vil inneholde unike verdier og kan brukes ikke bare til å vise «delta»-data (data lagt til/endret siden slutten av forrige utvekslingsøkt), men også for enkelt og effektivt å dele dem opp i sider.

Den siste foreslåtte metoden for å minimere mengden data som overføres innenfor rammen av replikering på høyt nivå virker for meg den mest optimale og universelle. La oss se på det mer detaljert.

Sende data ved hjelp av en radversjonsteller

Implementering av server/masterdel

I MS SQL Server er det en spesiell kolonnetype for å implementere denne tilnærmingen - rowversion. Hver database har en teller som øker med én hver gang en post legges til/endres i en tabell som har en kolonne som rowversion. Verdien til denne telleren blir automatisk tilordnet feltet i denne kolonnen i den lagte/endrede posten. Tarantool DBMS har ikke en lignende innebygd mekanisme. Men i Tarantool er det ikke vanskelig å implementere det manuelt. La oss se på hvordan dette gjøres.

Først en liten terminologi: tabeller i Tarantool kalles mellomrom, og poster kalles tupler. I Tarantool kan du lage sekvenser. Sekvenser er ikke annet enn navngitte generatorer av ordnede heltallsverdier. De. dette er akkurat det vi trenger for våre formål. Nedenfor vil vi lage en slik sekvens.

Før du utfører noen databaseoperasjon i Tarantool, må du kjøre følgende kommando:

box.cfg{}

Som et resultat vil Tarantool begynne å skrive database øyeblikksbilder og transaksjonslogger til gjeldende katalog.

La oss lage en sekvens row_version:

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

alternativ if_not_exists lar opprettelsesskriptet kjøres flere ganger: hvis objektet eksisterer, vil ikke Tarantool prøve å lage det igjen. Dette alternativet vil bli brukt i alle påfølgende DDL-kommandoer.

La oss lage et rom 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 setter vi navnet på rommet (goods), feltnavn og deres typer.

Automatisk inkrementerende felt i Tarantool lages også ved hjelp av sekvenser. La oss lage en auto-inkrementerende primærnøkkel etter 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 støtter flere typer indekser. De mest brukte indeksene er TREE- og HASH-typer, som er basert på strukturer som tilsvarer navnet. TREE er den mest allsidige indekstypen. Den lar deg hente data på en organisert måte. Men for likestillingsvalg er HASH mer egnet. Følgelig er det tilrådelig å bruke HASH for primærnøkkelen (som er det vi gjorde).

For å bruke kolonnen row_ver for å overføre endrede data, må du binde sekvensverdier til feltene i denne kolonnen row_ver. Men i motsetning til primærnøkkelen, kolonnefeltets verdi row_ver bør øke med én ikke bare når du legger til nye poster, men også når du endrer eksisterende. Du kan bruke triggere til dette. Tarantool har to typer plassutløsere: before_replace и on_replace. Triggere utløses hver gang dataene i rommet endres (for hver tuppel som påvirkes av endringene, startes en triggerfunksjon). I motsetning til on_replace, before_replace-triggere lar deg endre dataene til tuppelen som triggeren utføres for. Følgelig passer den siste typen triggere oss.

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 utløser erstatter feltverdien row_ver lagret tuppel til neste verdi i sekvensen row_version.

For å kunne trekke ut data fra verdensrommet goods etter kolonne row_ver, la oss lage en indeks:

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

Indekstype - tre (TREE), fordi vi må trekke ut dataene i stigende rekkefølge av verdiene i kolonnen row_ver.

La oss legge til noen data til plassen:

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 feltet er en automatisk økende teller; vi passerer null i stedet. Tarantool vil automatisk erstatte den neste verdien. På samme måte som verdien av kolonnefeltene row_ver du kan passere null - eller ikke spesifisere verdien i det hele tatt, fordi denne kolonnen opptar den siste posisjonen i rommet.

La oss sjekke innsettingsresultatet:

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 fylles første og siste felt ut automatisk. Nå blir det enkelt å skrive en funksjon for side-for-side opplasting av plassendringer 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

Funksjonen tar verdien som parameter row_ver, fra hvilken det er nødvendig å laste ned endringer, og returnerer en del av de endrede dataene.

Datasampling i Tarantool gjøres gjennom indekser. Funksjon get_goods bruker en iterator etter indeks row_ver for å motta endrede data. Iteratortypen er GT (Større enn, større enn). Dette betyr at iteratoren vil sekvensielt krysse indeksverdiene fra den godkjente nøkkelen (feltverdi row_ver).

Iteratoren returnerer tupler. For senere å kunne overføre data via HTTP, er det nødvendig å konvertere tuplene til en struktur som er praktisk for påfølgende serialisering. Eksemplet bruker standardfunksjonen for dette tomap. I stedet for å bruke tomap du kan skrive din egen funksjon. For eksempel vil vi kanskje gi nytt navn til et felt name, ikke passerer feltet code og legg til 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 på utdataene (antall poster i én del) bestemmes av variabelen page_size. I eksemplet er verdien page_size er 5. I et ekte program betyr vanligvis sidestørrelsen mer. Det avhenger av den gjennomsnittlige størrelsen på romtupelen. Den optimale sidestørrelsen kan bestemmes empirisk ved å måle dataoverføringstid. Jo større sidestørrelsen er, desto mindre antall rundturer mellom avsender- og mottakersiden. På denne måten kan du redusere den totale tiden for nedlasting av endringer. Men hvis sidestørrelsen er for stor, vil vi bruke for lang tid på serveren for å serialisere prøven. Som et resultat kan det oppstå forsinkelser i behandlingen av andre forespørsler som kommer til serveren. Parameter page_size kan lastes fra konfigurasjonsfilen. For hver overført plass kan du angi sin egen verdi. For de fleste mellomrom kan imidlertid standardverdien (for eksempel 100) være passende.

La oss utføre funksjonen 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
...

La oss ta feltverdien row_ver fra siste linje og kall opp funksjonen igjen:

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
...

Igjen:

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

Som du kan se, når den brukes på denne måten, returnerer funksjonen alle plassposter side for side goods. Den siste siden etterfølges av et tomt utvalg.

La oss gjøre endringer i plassen:

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

Vi har endret feltverdien name for én oppføring og lagt til to nye oppføringer.

La oss gjenta det siste funksjonsanropet:

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
...

Funksjonen returnerte de endrede og tilføyde postene. Så funksjonen get_goods lar deg motta data som har endret seg siden siste samtale, som er grunnlaget for replikeringsmetoden som vurderes.

Vi vil la utstedelse av resultater via HTTP i form av JSON være utenfor rammen av denne artikkelen. Du kan lese om dette her: https://habr.com/ru/company/mailru/blog/272141/

Implementering av klient/slave delen

La oss se på hvordan mottakersidens implementering ser ut. La oss lage en plass på mottakersiden for å lagre de nedlastede dataene:

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
})

Strukturen til rommet ligner strukturen til rommet i kilden. Men siden vi ikke skal sende de mottatte dataene noe annet sted, kolonnen row_ver er ikke i mottakerens plass. I felt id kildeidentifikatorer vil bli registrert. På mottakersiden er det derfor ikke nødvendig å gjøre den automatisk inkrementerende.

I tillegg trenger vi en plass for å lagre verdier 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 lastet plass (felt space_name) vil vi lagre den sist innlastede verdien her row_ver (felt value). Kolonnen fungerer som primærnøkkel space_name.

La oss lage en funksjon for å laste plassdata goods via HTTP. For å gjøre dette trenger vi et bibliotek som implementerer en HTTP-klient. Følgende linje laster biblioteket og instansierer HTTP-klienten:

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

Vi trenger også et bibliotek for json-deserialisering:

local json = require('json')

Dette er nok til å lage en datainnlastingsfunksjon:

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

Funksjonen utfører en HTTP-forespørsel til url-adressen og sender den row_ver som en parameter og returnerer det deserialiserte resultatet av forespørselen.

Funksjonen for å lagre mottatte data ser slik ut:

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

Syklus for å lagre data til plass goods plassert i en transaksjon (funksjonen brukes til dette box.atomic) for å redusere antall diskoperasjoner.

Til slutt den lokale romsynkroniseringsfunksjonen goods med en kilde kan du implementere det slik:

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 leser vi den tidligere lagrede verdien row_ver for plass goods. Hvis det mangler (første utvekslingsøkt), så tar vi det som row_ver null. Neste i syklusen utfører vi en side-for-side-nedlasting av de endrede dataene fra kilden på den angitte url. Ved hver iterasjon lagrer vi de mottatte dataene til det aktuelle lokale området og oppdaterer verdien row_ver (i verdensrommet row_ver og i variabelen row_ver) - ta verdien row_ver fra den siste linjen med innlastede data.

For å beskytte mot utilsiktet looping (i tilfelle feil i programmet), løkken while kan erstattes av for:

for _ = 1, max_req do ...

Som et resultat av å utføre funksjonen sync_goods rom goods mottakeren vil inneholde de nyeste versjonene av alle romposter goods i kilden.

Det er klart at sletting av data ikke kan kringkastes på denne måten. Hvis et slikt behov eksisterer, kan du bruke et slettemerke. Legg til plass goods boolsk felt is_deleted og i stedet for å slette en post fysisk, bruker vi logisk sletting - vi setter feltverdien is_deleted til mening true. Noen ganger i stedet for et boolsk felt is_deleted det er mer praktisk å bruke feltet deleted, som lagrer dato-klokkeslett for den logiske slettingen av posten. Etter å ha utført en logisk sletting, vil posten merket for sletting bli overført fra kilden til destinasjonen (i henhold til logikken diskutert ovenfor).

sekvens row_ver kan brukes til å overføre data fra andre rom: det er ikke nødvendig å lage en separat sekvens for hvert overført rom.

Vi så på en effektiv måte for datareplikering på høyt nivå i applikasjoner som bruker Tarantool DBMS.

Funn

1. Tarantool DBMS er et attraktivt, lovende produkt for å lage høybelastningsapplikasjoner.
2. Datareplikering på høyt nivå har en rekke fordeler fremfor replikering på lavt nivå.
3. Høynivåreplikeringsmetoden omtalt i artikkelen lar deg minimere mengden overførte data ved å overføre bare de postene som har endret seg siden siste utvekslingsøkt.

Kilde: www.habr.com