Cześć, zajmuję się tworzeniem aplikacji dla baz danych. — to platforma opracowana przez Mail.ru Group, która łączy w sobie wydajną bazę danych oraz serwer aplikacji w języku Lua. Wysoka prędkość rozwiązań opartych na Tarantool osiągana jest między innymi dzięki wsparciu trybu in-memory bazy danych oraz możliwości wykonywania logiki biznesowej aplikacji w tej samej przestrzeni adresowej co dane. Zapewnia to trwałość danych przy wykorzystaniu transakcji ACID (na dysku prowadzone jest dziennik WAL). Tarantool ma wbudowaną obsługę replikacji i shardingu. Od wersji 2.1 wspierane są zapytania w języku SQL. Tarantool ma otwarty kod źródłowy i jest dystrybuowany na zasadzie licencji Simplified BSD. Istnieje również komercyjna wersja Enterprise.
Poczuj moc! (…aka ciesz się wydajnością)
Wszystko to sprawia, że Tarantool jest atrakcyjną platformą do tworzenia wysoko obciążonych aplikacji współpracujących z bazą danych. W takich aplikacjach często pojawia się potrzeba replikacji danych.
Jak wspomniano wcześniej, w Tarantool istnieje wbudowana replikacja danych. Jej zasada działania polega na sekwencyjnym wykonywaniu wszystkich transakcji zawartych w dzienniku mastera (WAL) na replikach. Zwykle taka replikacja (nazywana dalej niskopoziomową) wykorzystywana jest do zapewnienia odporności aplikacji na awarie oraz/lub do rozkładania obciążenia na odczyty między węzłami klastra.
Rys. 1. Replikacja wewnątrz klastra
Przykładem alternatywnego scenariusza może być przesyłanie danych stworzonych w jednej bazie danych do innej bazy w celu przetwarzania/monitorowania. W ostatnim przypadku bardziej wygodnym rozwiązaniem może okazać się wykorzystanie wysokopoziomowej replikacji — replikacji danych na poziomie logiki biznesowej aplikacji. Tzn. nie wykorzystujemy gotowego rozwiązania, wbudowanego w bazę danych, a sami realizujemy replikację w naszej aplikacji. Takie podejście ma zarówno zalety, jak i wady. Wymieńmy plusy.
1. Oszczędność transferu:
- można przekazywać nie wszystkie dane, a tylko ich wybrane fragmenty (na przykład, można przekazywać tylko niektóre tabele, wybrane ich kolumny lub rekordy, które spełniają określone kryterium);
- W przeciwieństwie do niskopoziomowej replikacji, która odbywa się nieprzerwanie w trybie asynchronicznym (wdrożonym w bieżącej wersji Tarantool — 1.10) lub synchronicznym (wdrożonym w przyszłych wersjach Tarantool), replikację wysokopoziomową można realizować w sesjach (tj. aplikacja najpierw synchronizuje dane — sesja wymiany danych, następnie następuje przerwa w replikacji, po której odbywa się następna sesja wymiany itp.);
- jeśli zapis zmienił się wielokrotnie, można przekazać tylko jego ostatnią wersję (w przeciwieństwie do niskopoziomowej replikacji, w której na replikach zostaną kolejno odtworzone wszystkie zmiany dokonane na masterze).
2. Brak skomplikowanych implementacji wymiany przez HTTP, co umożliwia synchronizację zdalnych baz danych.
Rys. 2. Replikacja przez HTTP
3. Struktury baz danych, między którymi są przesyłane dane, nie muszą być identyczne (co więcej, w ogólnym przypadku możliwe jest nawet użycie różnych DBMS, języków programowania, platform itp.).
Rys. 3. Replikacja w heterogenicznych systemach
Minusem jest to, że programowanie jest w średniej bardziej złożone/ kosztowne niż konfiguracja, a zamiast dostosowywania wbudowanej funkcjonalności, trzeba zrealizować własną.
Jeżeli w Waszej sytuacji przedstawione zalety mają decydujące znaczenie (lub są niezbędnym warunkiem), warto zastosować replikację wysokopoziomową. Rozważmy kilka sposobów wdrażania replikacji wysokopoziomowej danych w bazie danych Tarantool.
Minimalizacja ruchu
Zatem jednym z zalet replikacji wysokopoziomowej jest oszczędność ruchu. Aby ta zaleta mogła się w pełni ujawnić, należy zminimalizować ilość danych przesyłanych podczas każdej sesji wymiany. Oczywiście, nie można zapominać, że na końcu sesji odbiornik danych musi być zsynchronizowany z źródłem (co najmniej w zakresie danych biorących udział w replikacji).
Jak więc zminimalizować ilość danych przesyłanych przy replikacji wysokopoziomowej? Rozwiązaniem „z głowy” może być selekcja danych według daty-czasu. W tym celu można wykorzystać już istniejące w tabeli pole daty-czasu (jeśli istnieje). Na przykład dokument „zamówienie” może mieć pole „wymagana data realizacji zamówienia” — czas_dostawy. Problem tego rozwiązania polega na tym, że wartości w tym polu nie muszą być uporządkowane zgodnie z kolejnością składania zamówień. W związku z tym nie możemy zapamiętać maksymalnej wartości pola czas_dostawy, przekazanej podczas poprzedniej sesji wymiany, i podczas następnej sesji wymiany wybrać wszystkich rekordów z wyższą wartością pola czas_dostawy. W międzyczasie mogły zostać dodane rekordy z mniejszą wartością pola czas_dostawy. Ponadto zamówienie mogło ulec zmianom, które jednak nie wpłynęły na pole czas_dostawy. W obu przypadkach zmiany nie będą przesyłane z źródła do odbiorcy. Aby rozwiązać te problemy, będziemy musieli przesyłać dane "na zakładkę". To znaczy, że podczas każdej sesji wymiany będziemy przesyłać wszystkie dane z wartością pola czas_dostawy, przekraczającą pewny moment w przeszłości (na przykład N godzin od chwili obecnej). Jednak oczywiste jest, że dla dużych systemów takie podejście jest znacznie nieefektywne i może zniweczyć oszczędności w ruchu sieciowym, do których dążymy. Co więcej, w przesyłanej tabeli może nie być pola związanego z datą i czasem.
Inne rozwiązanie, bardziej skomplikowane pod względem realizacji, polega na potwierdzeniu odbioru danych. W takim przypadku podczas każdej sesji wymiany przesyłane są wszystkie dane, których odbiór nie został potwierdzony przez odbiorcę. Aby to zrealizować, konieczne będzie dodanie do tabeli-źródła kolumny logicznej (na przykład, is_transferred). Jeśli odbiorca potwierdza odbiór rekordu, odpowiednie pole przyjmuje wartość true, po czym rekord nie bierze już udziału w wymianach. Taka wersja realizacji ma następujące wady. Po pierwsze, dla każdego przesyłanego rekordu należy wygenerować i wysłać potwierdzenie. Mówiąc wprost, może to być porównywalne z podwojeniem ilości przesyłanych danych i doprowadzić do podwojenia liczby rund. Po drugie, nie ma możliwości przesyłania tego samego rekordu do kilku odbiorców (pierwszy odbiorca, który otrzyma, potwierdzi odbiór za siebie i za wszystkich innych).
Metoda wolna od powyższych wad polega na dodaniu do przesyłanej tabeli kolumny do śledzenia zmian jej wierszy. Taka kolumna może mieć typ data-czas i powinna być ustalana/aktualizowana przez aplikację na bieżący czas za każdym razem przy dodawaniu/zamianie rekordów (atomowo z dodawaniem/zmianą). Jako przykład nazwijmy tę kolumnę aktualizacja_czasu. Zachowując maksymalną wartość pola tej kolumny dla przesyłanych rekordów, będziemy mogli rozpocząć następna sesję wymiany od tej wartości (wybrać rekordy z wartością pola aktualizacja_czasu, przewyższającą wcześniej zapisaną wartość). Problem związany z ostatnim podejściem polega na tym, że zmiany danych mogą zachodzić w trybie wsadowym. W wyniku tego wartości pól w kolumnie aktualizacja_czasu mogą być nieunikalne. Tak więc ta kolumna nie może być używana do stronicowania danych. Do stronicowania danych będziemy musieli stworzyć dodatkowe mechanizmy, które najprawdopodobniej będą miały bardzo niską wydajność (na przykład pobieranie z bazy danych wszystkich rekordów z wartością aktualizacja_czasu wyższą od zadanej i wydanie określonej ilości rekordów, zaczynając od pewnego przesunięcia od początku próbki).
Można poprawić efektywność przesyłania danych, nieco udoskonalając poprzednie podejście. W tym celu jako wartości pól kolumny do śledzenia zmian będziemy używać typu numerycznego (długie całkowite). Nazwijmy kolumnę wersja_wiersza. Wartość pola tej kolumny powinna być nadal ustalana/aktualizowana za każdym razem przy tworzeniu/zamianie rekordu. Ale w tym przypadku polu zostanie przypisana nie bieżąca data-czas, lecz wartość pewnego licznika, zwiększonego o jeden. W rezultacie kolumna wersja_wiersza będzie zawierać unikalne wartości i będzie mogła być używana nie tylko do wydawania "delta" danych (danych, które zostały dodane/zmienione po zakończeniu poprzedniej sesji wymiany), ale także do prostego i efektywnego rozdzielania ich na strony.
Ostatnio zaproponowany sposób minimalizacji ilości przesyłanych danych w ramach wysokopoziomowej replikacji wydaje mi się najbardziej optymalny i uniwersalny. Zatrzymajmy się na nim dokładniej.
Przesyłanie danych z użyciem licznika wersji wierszy
Realizacja części serwerowej/master
W MS SQL Server do realizacji tego podejścia istnieje specjalny typ kolumny — rowversion. Każda baza danych ma licznik, który zwiększa się o jeden za każdym razem, gdy dodawany/zmieniany jest rekord w tabeli zawierającej kolumnę typu rowversion. Wartość tego licznika jest automatycznie przypisywana do pola tej kolumny w dodawanym/zmienianym rekordzie. SDB Tarantool nie ma odpowiedniego wbudowanego mechanizmu, jednak w Tarantool można to łatwo zaimplementować ręcznie. Zobaczmy, jak to zrobić.
Na początek trochę terminologii: tabele w Tarantool nazywane są spaces (przestrzeniami), a rekordy — krotkami (krotka). W Tarantool można tworzyć sekwencje (sequence). Sekwencje to nic innego jak nazwane generatory uporządkowanych wartości liczb całkowitych. Tzn. to dokładnie to, co potrzebujemy do naszych celów. Poniżej stworzymy taką sekwencję.
Zanim wykonasz jakąkolwiek operację z bazą danych w Tarantool, musisz wykonać następującą komendę:
box.cfg{}W rezultacie Tarantool zacznie zapisywać w bieżącym katalogu zrzuty bazy danych (snapshot) oraz dziennik transakcji.
Stwórzmy sekwencję row_version:
box.schema.sequence.create('row_version',
{ if_not_exists = true }) Opcja if_not_exists pozwala na wielokrotne wykonanie skryptu tworzenia: jeśli obiekt istnieje, Tarantool nie będzie próbował go tworzyć ponownie. Ta opcja będzie używana we wszystkich kolejnych poleceniach DDL.
Stwórzmy przestrzeń dla przykładu.
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
}) Tutaj określiliśmy nazwę przestrzeni (goods), nazwy pól i ich typy.
Pola z autoincrementem w Tarantool również są tworzone za pomocą sekwencji. Stwórzmy klucz główny autoincrementowy według pola 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 obsługuje kilka typów indeksów. Najczęściej używane to indeksy typu TREE i HASH, które opierają się na odpowiednich strukturach. TREE to najbardziej uniwersalny typ indeksu. Umożliwia on wydobycie danych w uporządkowany sposób. Natomiast do wyborów na podstawie równości lepiej nadaje się HASH. Dlatego do klucza głównego sensowne jest użycie HASH (co właśnie zrobiliśmy).
Aby użyć kolumny wersja_wiersza do przesyłania zmienionych danych, należy powiązać z polami tej kolumny wartości sekwencji wersja_wiersza. Jednak w przeciwieństwie do klucza głównego wartość pola kolumny wersja_wiersza powinna wzrastać o jeden nie tylko przy dodawaniu nowych rekordów, ale także przy modyfikacji istniejących. W tym celu można użyć wyzwalaczy. W Tarantool istnieją dwa typy wyzwalaczy dla przestrzeni: before_replace i on_replace. Wyzwalacze uruchamiają się przy każdej zmianie danych w przestrzeni (dla każdego krotki dotkniętego zmianami uruchamiana jest funkcja wyzwalacza). W przeciwieństwie do on_replace, before_replace-wyzwalacze pozwalają modyfikować dane krotki, dla której wykonywany jest wyzwalacz. Dlatego pasuje nam ostatni typ wyzwalaczy.
box.space.goods:before_replace(function(old, new)
return box.tuple.new({new[1], new[2], new[3],
box.sequence.row_version:next()})
end) Podany wyzwalacz zmienia wartość pola wersja_wiersza przechowywanej krotki na kolejne wartość sekwencji row_version.
Aby można było wydobywać dane z przestrzeni goods na podstawie kolumny wersja_wiersza, stworzymy indeks:
box.space.goods:create_index('row_ver', {
parts = { 'row_ver' },
unique = true,
type = 'TREE',
if_not_exists = true
}) Typ indeksu — drzewo (TREE), ponieważ będziemy potrzebować wydobywać dane w porządku rosnącym wartości w kolumnie wersja_wiersza.
Dodamy do przestrzeni kilka danych:
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} Ponieważ pierwsze pole to licznik autoincrement, przekazujemy zamiast niego nil. Tarantool automatycznie wstawi wartość kolejnego. W podobny sposób jako wartości pól kolumny wersja_wiersza można przekazać nil — lub w ogóle nie wskazywać wartości, ponieważ ta kolumna zajmuje ostatnią pozycję w przestrzeni.
Sprawdźmy wynik wstawienia:
tarantool> box.space.goods:select()
---
- - [1, 'długopis', 123, 1]
- [2, 'ołówek', 321, 2]
- [3, 'pędzel', 100, 3]
- [4, 'akwarela', 456, 4]
- [5, 'album', 101, 5]
- [6, 'zeszyt', 800, 6]
- [7, 'gumka', 531, 7]
- [8, 'linijka', 135, 8]
... Jak widać, pierwsze i ostatnie pole wypełniły się automatycznie. Teraz łatwo będzie napisać funkcję do paginacji zmian w przestrzeni. 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 Funkcja przyjmuje jako parametr wartość wersja_wiersza, od której należy rozpocząć wyodrębnianie zmian i zwraca porcję zmienionych danych.
Wyciąganie danych w Tarantool odbywa się przez indeksy. Funkcja get_goods używa iteratora po indeksie wersja_wiersza do pozyskiwania zmienionych danych. Typ iteratora — GT (Greater Than, większe niż). Oznacza to, że iterator będzie kolejno przechodził wartości indeksu, zaczynając od podanego klucza (wartości pola wersja_wiersza).
Iterator zwraca krotki. Aby następnie móc przesłać dane przez HTTP, należy przekształcić krotki w strukturę wygodną do późniejszej serializacji. W przykładzie do tego używana jest standardowa funkcja tomap. Zamiast używać tomap można napisać własną funkcję. Na przykład możemy chcieć zmienić nazwę pola name, nie przesyłać pola code i dodać pole komentarz:
local function unflatten_goods(tuple)
local obj = {}
obj.id = tuple.id
obj.goods_name = tuple.name
obj.comment = 'jakiś komentarz'
obj.row_ver = tuple.row_ver
return obj
end Rozmiar strony wydawanych danych (liczba rekordów w jednej porcji) określany jest przez zmienną page_size. W przykładzie wartość page_size wynosi 5. W rzeczywistym programie rozmiar strony ma zwykle większe znaczenie. Zależy od średniego rozmiaru krotki przestrzeni. Optymalny rozmiar strony można dobrać empirycznie, mierząc czas transferu danych. Im większy rozmiar strony, tym mniejsza liczba rund pomiędzy stroną wysyłającą a odbierającą. Można w ten sposób skrócić całkowity czas przesyłania zmian. Jednak przy zbyt dużym rozmiarze strony zbyt długo zajmujemy serwer serializacją zbioru. W rezultacie mogą wystąpić opóźnienia w przetwarzaniu innych żądań, które dotarły do serwera. Parametr page_size można załadować z pliku konfiguracyjnego. Dla każdej przesyłanej przestrzeni można ustawić swoją wartość. Przy tym dla większości przestrzeni odpowiednia może być wartość domyślna (na przykład 100).
Wykonajmy funkcję get_goods:
tarantool> get_goods(0)
---
- - row_ver: 1
code: 123
name: długopis
id: 1
- row_ver: 2
code: 321
name: ołówek
id: 2
- row_ver: 3
code: 100
name: pędzel
id: 3
- row_ver: 4
code: 456
name: akwarela
id: 4
- row_ver: 5
code: 101
name: album
id: 5
... Weźmy wartość pola wersja_wiersza z ostatniego wiersza i ponownie wywołajmy funkcję:
tarantool> get_goods(5)
---
- - row_ver: 6
code: 800
name: zeszyt
id: 6
- row_ver: 7
code: 531
name: gumka
id: 7
- row_ver: 8
code: 135
name: linijka
id: 8
...I jeszcze raz:
tarantool> get_goods(8)
---
- []
... Jak widzimy, przy takim użyciu funkcja stronnicowo zwraca wszystkie zapisy przestrzeni goods. Za ostatnią stroną następuje pusta kwerenda.
Wprowadźmy zmiany w przestrzeni:
box.space.goods:update(4, {{'=', 6, 'zeszyt'}})
box.space.goods:insert{nil, 'klips', 234}
box.space.goods:insert{nil, 'folder', 432} Zmieniliśmy wartość pola name dla jednego rekordu i dodaliśmy dwa nowe rekordy.
Powtórzmy ostatnie wywołanie funkcji:
tarantool> get_goods(8)
---
- - row_ver: 9
code: 800
name: zeszyt
id: 6
- row_ver: 10
code: 234
name: klips
id: 9
- row_ver: 11
code: 432
name: folder
id: 10
... Funkcja zwróciła zmienione oraz dodane zapisy. W ten sposób funkcja get_goods umożliwia uzyskiwanie danych, które zmieniły się od chwili ostatniego wywołania, co stanowi podstawę omawianego sposobu replikacji.
Zostawimy wydawanie wyników przez HTTP w formie JSON poza zakresem niniejszego artykułu. Można o tym przeczytać tutaj:
Implementacja części klienckiej/slave
Rozważmy, jak wygląda implementacja strony odbierającej. Stworzymy na stronie odbierającej przestrzeń do przechowywania załadowanych danych:
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
}) Struktura przestrzeni przypomina strukturę przestrzeni w źródle. Ale ponieważ nie zamierzamy przesyłać uzyskanych danych gdzie indziej, kolumna wersja_wiersza w przestrzeni odbiorcy jest nieobecna. W polu id będą zapisywane identyfikatory źródła. Dlatego po stronie odbiorcy nie ma potrzeby, aby był autoinkrementalny.
Ponadto potrzebujemy przestrzeni do przechowywania wartości wersja_wiersza:
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
}) Dla każdej ładowanej przestrzeni (pole space_name) będziemy tu przechowywać ostatnią załadowaną wartość wersja_wiersza (pole value). Jako klucz główny występuje kolumna space_name.
Stworzymy funkcję do ładowania danych przestrzeni goods przez HTTP. W tym celu potrzebujemy biblioteki, która realizuje klienta HTTP. Następna linia ładuje bibliotekę i tworzy instancję klienta HTTP:
local http_client = require('http.client').new()Potrzebujemy również biblioteki do deserializacji json:
local json = require('json')To wystarczy do stworzenia funkcji ładującej dane:
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 Funkcja wykonuje żądanie HTTP pod adresem url, przekazuje w nim wersja_wiersza jako parametr i zwraca zdeserializowany wynik zapytania.
Funkcja zapisywania uzyskanych danych wygląda następująco:
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 Pętla zapisywania danych w przestrzeni goods umieszczona jest w transakcji (do tego używana jest funkcja box.atomic) aby zmniejszyć liczbę operacji z dyskiem.
Na końcu funkcję synchronizacji lokalnej przestrzeni goods ze źródłem można zaimplementować w ten sposób:
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 Najpierw odczytujemy wcześniej zapisane wartość wersja_wiersza dla przestrzeni goods. Jeśli go nie ma (pierwsza sesja wymiany), to bierzemy jako wersja_wiersza zero. Then, in the loop, we perform page-by-page loading of modified data from the source at the specified url. At each iteration, we save the retrieved data in the corresponding local space and update the value wersja_wiersza (in the space wersja_wiersza and in the variable wersja_wiersza) — we take the value wersja_wiersza from the last line of the loaded data.
To protect against accidental looping (in case of a program error), the loop while can be replaced with for:
for _ = 1, max_req do ... As a result of the function execution sync_goods the space goods in the receiver will contain the latest versions of all records from the space goods in the source.
It is obvious that this method cannot be used to transmit data deletion. If such a need exists, a deletion marker can be used. We add to the space goods a boolean field is_deleted and instead of physically deleting a record, we use logical deletion — we set the value of the field is_deleted na wartość true. Sometimes, instead of a boolean field is_deleted it is more convenient to use a field deletedthat stores the date-time of logical deletion of the record. After logical deletion is performed, the marked record for deletion will be transferred from the source to the receiver (according to the logic discussed above).
The sequence wersja_wiersza can be used to transfer data from other spaces: there is no need to create a separate sequence for each space being transferred.
We have considered an effective method of high-level data replication in applications using the Tarantool DBMS.
Wnioski
- The Tarantool DBMS is an attractive, promising product for creating high-load applications.
- High-level data replication has several advantages over low-level replication.
- The method of high-level replication discussed in the article allows minimizing the amount of data transmitted by only transferring those records that have changed since the last exchange session.
Źródło: habr.com
