Здравейте, занимавам се с разработка на приложения за СУБД. — това е платформа, разработена в Mail.ru Group, която съчетава високо производителна СУБД и сървър за приложения на Lua. Високата скорост на решенията, базирани на Tarantool, се постига частично благодарение на поддръжката на in-memory режим на СУБД и възможността за изпълнение на бизнес логиката на приложението в едно адресно пространство с данните. При това се осигурява постоянство на данните с използването на ACID транзакции (на диска се води WAL журнал). Tarantool има вградена поддръжка за репликация и шардиринг. Започвайки от версия 2.1, се поддържат запитвания на SQL. Tarantool има отворен код и се разпространява под лиценз Simplified BSD. Има и комерсиална Enterprise версия.
Почувствайте силата! (…или се насладете на производителността)
Всичко това прави Tarantool привлекателна платформа за създаване на приложения с високо натоварване, работещи с БД. В такива приложения често възниква необходимост от репликация на данни.
Както беше споменато по-горе, в Tarantool има вградена репликация на данни. Принципът на работа е основан на последователно изпълнение на всичките транзакции, съдържащи се в журнала на мастера (WAL) на репликите. Обикновено такава репликация (наречена по-долу нисконивова) се използва за осигуряване на отказоустойчивост на приложението и/или за разпределение на натоварването при четене между нодовете на клъстера.
Рис. 1. Репликация в рамките на клъстера
Пример за алтернативен сценарий може да бъде прехвърлянето на данни, създадени в една БД, в друга БД за обработка/мониторинг. В последния случай може да се окаже, че по-удобно решение е използването на високонивова репликация — репликация на данни на ниво бизнес логика на приложението. Т.е. не използваме готово решение, вградено в СУБД, а сами реализираме репликация във разрабатываното от нас приложение. Този подход има както предимства, така и недостатъци. Нека изброим плюсовете.
1. Икономия на трафик:
- може да се предават не всички данни, а само част от тях (например, може да се предават само някои таблици, определени колони или записи, отговарящи на определен критерий);
- в отличие от нисшестого репликирования, което се извършва без прекъсване в асинхронен (реализирано в текущата версия Tarantool — 1.10) или синхронен (ще бъде реализирано в бъдещи версии Tarantool) стил, висшето репликиране може да се извършва чрез сесии (т.е. приложението първо извършва синхронизация на данни — сесия за обмен на данни, след това следва пауза в репликирането, след която следва следваща сесия за обмен и т.н.);
- ако записа е променен няколко пъти, може да се предаде само последната му версия (в отличие от нисшестото репликиране, при което на репликите последователно ще се възпроизведат всички промени, направени на мастера).
2. Липсват трудности при реализиране на обмена по HTTP, което позволява синхронизиране на отдалечени БД.
Рис. 2. Репликация по HTTP
3. Структурите на БД, между които се предават данни, не е необходимо да са еднакви (по-скоро, в общия случай, дори е възможно да се използват различни СУБД, програмни езици, платформи и т.н.).
Рис. 3. Репликация в хетерогенни системи
Недостатъкът е, че средно програмирането е по-сложно/скъпо от конфигурирането, и вместо да се настройва вградения функционал, ще трябва да се реализира собствен.
Ако в ситуацията ви посочените предимства играят решаваща роля (или са необходимо условие), то има смисъл да се използва висшето репликиране. Да разгледаме няколко начина за реализиране на висшето репликиране на данни в СУБД Tarantool.
Минимизиране на трафика
И така, едно от предимствата на висшето репликиране е икономията на трафик. За да се прояви това предимство в пълнота, е необходимо да се минимизира количеството данни, предавани при всяка сесия на обмен. Разбира се, не бива да забравяме, че в края на сесията приемникът на данни трябва да бъде синхронизиран с източника (поне по онези данни, които участва в репликирането).
Как да минимизираме количеството данни, предавани при висшето репликиране? Решението "в лоб" може да бъде отбор на данни по дата-време. За целта може да се използва вече наличното в таблицата поле за дата-време (ако такова има). Например, документа "заказ" може да има поле "изискуемо време за изпълнение на поръчката" — време за доставка. Проблемата на това решение е, че стойностите в това поле не е задължително да бъдат разположени в реда, съответстващ на създаването на поръчките. По този начин не можем да запомним максималната стойност на полето време за доставка, предадено при предишната сесия за обмен, и при следващата сесия за обмен да изберем всички записи с по-висока стойност на полето време за доставка. В интервала между сесиите за обмен могат да бъдат добавени записи с по-ниска стойност на полето време за доставка. Също така, поръчката може да е претърпяла промени, които обаче не са засегнали полето време за доставка. В двата случая промените няма да бъдат предадени от източника на получателя. За да решим тези проблеми, ще трябва да предаваме данни "внахлест". Т.е. при всяка сесия за обмен ще предаваме всички данни със стойност на полето време за доставка, надвишаваща определен момент в миналото (например, N часа от текущия момент). Въпреки това е очевидно, че за големи системи подобен подход е силно излишен и може да обезсили спестяването на трафик, към което се стремим. Освен това в предавана таблица може да няма поле, свързано с дата-време.
Друго решение, по-сложно от гледна точка на реализация, е потвърждаване на получаването на данни. В този случай при всяка сесия за обмен се предават всички данни, чието получаване не е потвърдено от получателя. За реализация ще е необходимо да добавим в източната таблица булева колона (например, is_transferred). Ако получателят потвърди получаването на записа, съответното поле приема стойност true, след което записът повече не участва в обмените. Такъв вариант на реализация има следните недостатъци. Първо, за всеки предаден запис е необходимо да се генерира и изпрати потвърждение. Грубо казано, това може да бъде сравнимо с удвояване на количеството предавани данни и да доведе до удвояване на броя на обиколките. На второ място, липсва възможност за изпращане на един и същи запис до няколко получатели (първият получил получател потвърдява получаването за себе си и за всички останали).
Метод, свободный от указанных выше недостатков, заключается в добавлении в передаваемую таблицу столбца для отслеживания изменений её строк. Этот столбец может иметь тип дата-время и должен задаваться/обновляться приложением на текущее время каждый раз при добавлении/изменении записей (атомарно с добавлением/изменением). Примером может служить столбец update_time. Сохранив максимальное значение поля этого столбца для переданных записей, мы сможем начать следующий сеанс обмена с этого значения (отобрать записи с полем update_time, превышающим ранее сохраненное значение). Проблема, связанная с последним подходом, заключается в том, что изменения данных могут происходить в пакетном режиме. В результате значения полей в столбце update_time могут не быть уникальными. Таким образом, этот столбец не может быть использован для порционной (постраничной) выдачи данных. Для постраничной выдачи данных придется изобретать дополнительные механизмы, которые, скорее всего, будут иметь очень низкую эффективность (например, извлечение из БД всех записей со значением update_time выше заданного и выдача определенного количества записей, начиная с некоторого смещения от начала выборки).
Можно повысить эффективность передачи данных, немного усовершенствовав предыдущий подход. Для этого в качестве значений полей столбца для отслеживания изменений будем использовать целочисленный тип (длинное целое). Назовем столбец row_ver. Значение поля этого столбца по-прежнему должно задаваться/обновляться каждый раз при создании/изменении записи. Но в этом случае полю будет присваиваться не текущее дата-время, а значение некоторого счетчика, увеличенного на единицу. В результате столбец row_ver будет содержать уникальные значения и сможет быть использован не только для выдачи «дельты» данных (данные, добавившиеся/изменившиеся после завершения предыдущего сеанса обмена), но и для простой и эффективной разбивки их на страницы.
Последний предложенный способ минимизации количества данных, передаваемых в рамках высокоуровневой репликации, представляется мне наиболее оптимальным и универсальным. Остановимся на нем подробнее.
Передача данных с использованием счетчика версий строк
Реализация на сървърната/майсторска част
В MS SQL Server за реализиране на подобен подход съществува специален тип колона — rowversion. Всяка БД има брояч, който се увеличава с единица всеки път, когато се добавя/променя запис в таблица с колона от тип rowversion. Стойността на този брояч автоматично се присвоява на полето в добавения/променения запис. СУБД Tarantool няма аналогичен вграден механизъм. Въпреки това, в Tarantool не е трудно да го реализираме ръчно. Нека разгледаме как се прави това.
Първо, малко терминология: таблиците в Tarantool се наричат спейсове (space), а записите — кортежи (tuple). В Tarantool можем да създаваме последователности (sequence). Последователностите представляват именовани генератори на подредени стойности на цели числа. Тоест, точно това, което ни е нужно за нашите цели. По-долу ще създадем такава последователност.
Преди да изпълните каквато и да е операция с базата данни в Tarantool, е необходимо да изпълните следната команда:
box.cfg{}В резултат на това Tarantool ще започне да записва в текущия каталог снимки на БД (snapshot) и журнал на транзакциите.
Ще създадем последователност row_version:
box.schema.sequence.create('row_version',
{ if_not_exists = true }) Опция if_not_exists позволява многократно изпълнение на скрипта за създаване: ако обектът съществува, Tarantool няма да се опита да го създаде отново. Тази опция ще се използва във всички последващи DDL команди.
Ще създадем спейс за пример.
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
}) Тук зададохме името на спейса (goods), имената на полетата и техните типове.
Автоинкрементните полета в Tarantool също се създават с помощта на последователности. Ще създадем автоинкрементен първичен ключ по полето ид:
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 поддържа няколко типа индекси. Най-често използваните са индексите от тип TREE и HASH, които отговарят на името на съответстващата структура. TREE е най-универсалният тип индекс. Той позволява извличане на данни в подреден вид. Но за избор по равенство HASH е по-подходящ. Съответно, за първичен ключ е целесъобразно да се използва HASH (както и направихме).
За да използвате колоната row_ver за предаване на променените данни, е необходимо да свържете на полетата на тази колона стойности на последователността row_ver. Но в отличие от първичния ключ, стойността на полето на колоната row_ver трябва да се увеличава с единица не само при добавяне на нови записи, но и при променяне на съществуващите. За това може да се използват тригери. В Tarantool има два типа тригери за пространства: before_replace и on_replace. Триггерите се задействат при всяка промяна на данни в пространството (за всеки кортеж, засегнат от промените, се задейства функцията на тригера). В отличие от on_replace, before_replace-триггерите позволяват модифициране на данните в кортежа, за който се изпълнява триггерът. Съответно, ни подхожда последният тип триггери.
box.space.goods:before_replace(function(old, new)
return box.tuple.new({new[1], new[2], new[3],
box.sequence.row_version:next()})
end) Представеният триггер заменя стойността на полето row_ver на запазения кортеж с следващата стойност на последователността. row_version.
За да можем да извличаме данни от пространството goods по колоната row_ver, ще създадем индекс:
box.space.goods:create_index('row_ver', {
parts = { 'row_ver' },
unique = true,
type = 'TREE',
if_not_exists = true
}) Типът на индекса е дърво (TREE), тъй като данните ще трябва да се извличат в нарастващ ред на стойностите в колоната. row_ver.
Ще добавим някои данни в пространството:
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} Тъй като първото поле е автоинкрементен брояч, предаваме вместо него nil. Tarantool автоматично ще подстави следващата стойност. По аналогичен начин като стойност на полетата на колоната row_ver може да се предаде nil — или да не се указва стойност въобще, тъй като тази колона заема последната позиция в пространството.
Нека проверим резултата от вмъкването:
tarantool> box.space.goods:select()
---
- - [1, 'писалка', 123, 1]
- [2, 'молив', 321, 2]
- [3, 'четка', 100, 3]
- [4, 'акварел', 456, 4]
- [5, 'албум', 101, 5]
- [6, 'тетрадка', 800, 6]
- [7, 'гума', 531, 7]
- [8, 'линейка', 135, 8]
... Как виждаме, първото и последното поле са били запълнени автоматично. Сега няма да е трудно да напишем функция за извличане на страниците с промени в пространството. 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 Функцията приема като параметър стойността row_ver, от която трябва да се извърши извличането на промените, и връща част от променените данни.
Извличането на данни в Tarantool се извършва чрез индекси. Функцията get_goods използва итератор по индекса row_ver за получаване на променените данни. Видът на итератора е GT (Greater Than, по-голямо от). Това означава, че итераторът ще извършва последователен обход на стойностите на индекса, започвайки от предадения ключ (стойността на полето row_ver).
Итераторът връща кортежи. За да можем след това да предадем данните по HTTP, е необходимо да извършим преобразуване на кортежите в структура, удобна за последващо сериализиране. В примера за това се използва стандартната функция tomap. Вместо да използваме tomap можем да напишем собствена функция. Например, можем да искаме да преименуваме полето име, да не предаваме полето code и да добавим полето comment:
local function unflatten_goods(tuple)
local obj = {}
obj.id = tuple.id
obj.goods_name = tuple.name
obj.comment = 'някакъв коментар'
obj.row_ver = tuple.row_ver
return obj
end Размерът на страницата на извлечените данни (броя записи в една порция) се определя от променливата page_size. В примера стойността page_size равно 5. В реалната програма размерът на страницата обикновено има по-голямо значение. Той зависи от средния размер на кортежа на спейса. Оптималният размер на страницата може да бъде определен емпирично, измервайки времето за предаване на данни. Колкото по-голям е размерът на страницата, толкова по-малко са обиколките между изпращащата и получаващата страна. Така може да се намали общото време за изтегляне на промените. Въпреки това, при твърде голям размер на страницата, ще заема прекалено много време на сървъра за сериализация на избора. В резултат може да се получат закъснения при обработката на други заявки, постъпили на сървера. Параметър page_size може да бъде изтеглен от конфигурационен файл. За всеки предаван спейс може да се зададе собствена стойност. В същото време за повечето спейсове може да се използва стойността по подразбиране (например, 100).
Изпълняваме функция get_goods:
tarantool> get_goods(0)
---
- - row_ver: 1
code: 123
name: chemikal
id: 1
- row_ver: 2
code: 321
name: оловка
id: 2
- row_ver: 3
code: 100
name: четка
id: 3
- row_ver: 4
code: 456
name: акварел
id: 4
- row_ver: 5
code: 101
name: албум
id: 5
... Нека вземем стойността на полето row_ver от последния ред и отново да извикаме функцията:
tarantool> get_goods(5)
---
- - row_ver: 6
code: 800
name: тефтер
id: 6
- row_ver: 7
code: 531
name: гума
id: 7
- row_ver: 8
code: 135
name: линийка
id: 8
...И още веднъж:
tarantool> get_goods(8)
---
- []
... Както виждаме, при такова използване функцията връща всичките записи на спейса на страници. goods. След последната страница следва празен избор.
Нека внесем промени в спейса:
box.space.goods:update(4, {{'=', 6, 'учебник'}})
box.space.goods:insert{nil, 'клип', 234}
box.space.goods:insert{nil, 'папка', 432} Променихме стойността на полето име за един запис и добавихме две нови записи.
Нека повторим последното извикване на функцията:
tarantool> get_goods(8)
---
- - row_ver: 9
code: 800
name: учебник
id: 6
- row_ver: 10
code: 234
name: клип
id: 9
- row_ver: 11
code: 432
name: папка
id: 10
... Функцията върна променените и новите записи. По този начин, функцията get_goods позволява да получавате данни, променили се от момента на последното ѝ извикване, което е основата на разглеждания метод на репликация.
Оставяме извеждането на резултати по HTTP в JSON формат извън рамките на настоящата статия. За това може да се прочете тук:
Реализация на клиентската/слейв част
Нека да видим как изглежда реализацията от страна на получателя. Създаваме на страната на получателя спейс за съхранение на заредените данни:
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
}) Структура на пространството напомня структурата на пространството в източника. Но тъй като не планираме да предаваме получените данни на друго място, колоната row_ver в пространството на получателя липсва. В полето ид ще се записват идентификаторите на източника. Така че от страната на приемника няма нужда да бъде автоматично увеличавано.
Освен това ще ни е необходима пространство за съхраняване на стойности 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
}) За всяко зареждано пространство (поле space_name) тук ще съхраняваме последната заредена стойност row_ver (поле стойност). Първичният ключ ще бъде колоната space_name.
Ще създадем функция за зареждане на данни от пространството goods чрез HTTP. За това ще ни е необходима библиотека, реализираща HTTP-клиент. Следващият ред зарежда библиотеката и създава инстанция на HTTP-клиента:
local http_client = require('http.client').new()Също така ще ни трябва библиотека за десериализация на json:
local json = require('json')Това е достатъчно за създаване на функция за зареждане на данни:
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 Функцията извършва HTTP-запитване по адреса url, предава в него row_ver като параметър и връща десериализирания резултат от запитването.
Функцията за запазване на получените данни изглежда по следния начин:
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 Цикълът за запазване на данни в пространството goods е поставен в транзакция (за това се използва функцията box.atomic) за намаляване на броя на операциите с диска.
Накрая, функцията за синхронизиране на локалното пространство goods с източника може да бъде реализирана така:
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 Първо извличаме предишната запазена стойност row_ver за пространството goods. Ако тя липсва (първа сесия на размяна), то взимаме като row_ver нула. След това в цикъл извършваме странично зареждане на променените данни от източника по указаното url. На всяка итерация съхраняваме получените данни в съответното локално пространство и актуализираме стойността row_ver (в пространството row_ver и в променливата row_ver) — вземаме стойността row_ver от последния ред на заредените данни.
За защита от случайно зацикляне (в случай на грешка в програмата) цикълът докато може да се замени с for:
for _ = 1, max_req do ... В резултат на изпълнението на функцията sync_goods пространство goods в приемника ще съдържа последните версии на всичките записи в пространството goods в източника.
Очевидно е, че по този начин не може да се транслира изтриването на данни. Ако такава необходимост съществува, може да се използва бележка за изтриване. Добавяме в пространството goods булево поле is_deleted и вместо физическо изтриване на записа използваме логическо изтриване — задаваме стойността на полето is_deleted на стойност true. Понякога вместо булево поле is_deleted е по-удобно да се използва поле изтрит, в което се съхранява дата-време на логическото изтриване на записа. След извършване на логическото изтриване, маркираният за изтриване запис ще бъде предаден от източника на приемника (съгласно разгледаната по-горе логика).
Последователността row_ver може да се използва за предаване на данни от други пространства: няма нужда от създаване на отделна последователност за всяко предавано пространство.
Разгледахме ефективен начин за високо ниво на репликация на данни в приложения, използващи СУБД Tarantool.
Изводи
- СУБД Tarantool е привлекателен, перспективен продукт за създаване на високо натоварени приложения.
- Високото ниво на репликация на данни има редица предимства в сравнение с ниското ниво на репликация.
- Разгледаният в статията метод на високо ниво на репликация позволява минимизиране на количеството предавани данни чрез предаване само на онези записи, които са се променили след последната сесия на обмен.
Източник: habr.com
