Альтернатива MapReduce при поиске в распределенной БД

Привет, меня зовут Сатбек, я работаю в команде Tarantool. Расскажу, как реализовать в шардированном кластере поиск, скорость которого не зависит от количества мастеров и объёма хранимых данных. Условно назову этот способ индексным слоем:

• Опишу общую схему построения поиска.

• Приведу пример реализации.

• Дам рекомендации по разработке.

В статье я последовательно реализую простой CRUD-сервис с шардированным хранилищем, а также поиск по данным. Это поможет вам лучше понять, когда возникают проблемы с поиском по распределённым данным и как их решает индексный слой.

Использовать будем БД Tarantool (версия ≥ 1.10), а также фреймворк для построения кластеров Tarantool-Cartridge (версия 2.7.0).

Для лучшего понимания желательно познакомиться с фреймворком Tarantool-Cartidge, модулем vshard, а также языком Lua, так как пример написан на нём.

Общее описание приложения

В качестве примера реализуем простой CRUD-сервис с одной таблицей user, шардированной по id. В Tarantool используется термин space, далее по тексту "спейс". Исходный код приложения находится здесь.

Искомые данные будут храниться в спейсе user в таком формате:

|id(uuid)|bucket_id(number)|name(string)|birthdate(number)|phone_number(string)|

Пользователи шардируются по id. Есть primary index по id и secondary index по name.

Как искать в таком кластере?

Поиск по id

-- router side
local bucket_id = vshard.router.bucket_id(user_id)
user = vshard.router.callrw(bucket_id, 'get_user_by_id', {user_id}, {timeout = 1})
-- storage side
local M = {}
function M.get_user_by_id(id)
    local user_t = get_user_tuple_by_id(id)
    if user_t == nil then
        return nil, M.user_not_exists_err:new('id=%s', id)
    end
    return user_tuple_to_output(user_t)
end

Для поиска по id используется стандартный для Tarantool подход. Мы вычисляем bucket_id на стороне роутера, чтобы узнать, на каком storage-узле лежат наши данные, а затем шлём туда запрос. В результате не более чем за два запроса получаем искомое.

Поиск по name

Далее реализуем поиск по name. Здесь не обойтись одним запросом: необходимо обойти каждый master-storage в нашем кластере, поскольку неизвестно, где лежат наши данные.

-- router side
local M = {}
function M.find_users_by_name(name)
    local storage_uris = cartridge_rpc.get_candidates('app.roles.storage', {leader_only = true})
    local res_by_uri, err = cartridge_pool.map_call('storage_api.get_users_by_name', {name}, {uri_list = storage_uris, timeout = M.vshard_timeout})

    if err ~= nil then
        return nil, err
    end

    local result = {}
    for _, res in pairs(res_by_uri) do
        for _, user in pairs(res) do
            table.insert(result, user)
        end
    end
    return result
end

-- storage side
local M = {}
function M.get_users_by_name(user_name)
    local yield_every = 100
    local count = 1

    local result = {}
    for _, t in box.space.user.index.name:pairs({user_name}, 'EQ') do
        if count % yield_every == 0 then
            fiber.yield()
        end

        count = count + 1
        table.insert(result, user_tuple_to_output(t))
    end

    return result
end

В этом случае запросов будет не меньше, чем количество мастеров в кластере. Однако такой подход допустим, если искомые данные точно лежат на нескольких мастерах и если таких мастеров немного. В противном случае для эффективного поиска необходим дополнительный индекс по полю name. В контексте поиска по «имени человека» описанные выше допущения вполне справедливы при достаточно большом объёме данных.

Расширяем поиск

Теперь предположим, что вы хотите искать пользователя и по номеру телефона. Для этого можно использовать описанный выше поиск по имени: делаем на storage дополнительный индекс и реализуем MapReduce для поиска. Схема будет работать, и её даже относительно просто реализовать, особенно с помощью модуля Tarantool/crud. Но всё же у неё есть существенный недостаток: поле «номер телефона», как правило, есть только в одной записи user. Соответственно, запись с искомым значением будет находиться только на одном мастере. Но в схеме с MapReduce мы будем вынуждены обойти все мастеры, и если у вас их две штуки в кластере, то придётся сделать два запроса на чтение, если 10 — то 10 запросов, и т.д. И всё это будут сетевые запросы, которые очень дороги по сравнению с чтением из индекса, да ещё и могут упасть. Возникает новая проблема — избыточные сетевые запросы при поиске по уникальным полям.

Схема индексного слоя

Трудность в том, что нам нужно найти уникальные данные, но мы не знаем, на каком мастере они лежат. Значит, нужно хранить об этом информацию. Пожертвовать памятью, чтобы уменьшить количество сетевых запросов. 

Создадим специальный спейс user_search_index:

|user_id(string-uuid)|bucket_id(unsigned)|data_hash(string)|data(any)|

Этот спейс и будет поисковым слоем. При записи данных мы будет вычислять data_hash от информации, используемой для поиска:

--один из множества вариантов, как это можно сделать
local msgpack = require('msgpack')
local digest = require('digest')

local data = {'phone-number', '202-555-0165'}
local data_hash = digest.md5_hex(msgpack.encode(data))

Спейс user_search_index шардируется по data_hash. И чтобы исключить коллизии, нужно хранить data. Здесь можно найти модуль приложения, реализующий логику построения хэшей:

local M = {}

local digest = require('digest')
local msgpack = require('msgpack')
M.cmp_data = {}

local Identifier = {}

function Identifier:new(data)
    local obj = {}
    obj.hash = digest.md5_hex(msgpack.encode(data))
    obj.data = data
    return obj
end

function M.phone_number(phone_number)
    local data = {'phone_number', phone_number}
    return Identifier:new(data)
end

function M.cmp_data.phone_number(data_one, data_two)
    return #data_one == 2 and #data_two == 2 and (data_one[1] == data_two[1]) and (data_one[2] == data_two[2] ~= nil)
end

return M

Read

При поиске:

• вычисляем хэш от поисковых данных;

• находим по нему данные из user_search_index;

• шлём запрос на storage и получаем данные по primary-key.

Пример кода для поиска по номеру телефона:

-- api.lua
local M = {}
function M.find_users_by_phone_number(phone_number)
    local user_ids, err = M.search_index.user_id.get_by_phone_number(phone_number)
    if err ~= nil then
        err = errors.wrap(err)
        return nil, err
    end

    return get_users_by_ids(user_ids)
end

-- search_api.lua
local M = {}
function M.user_id.get_by_phone_number(phone_number)
    local identifier = M.identifier.phone_number(phone_number)

    local bucket_id = M.vshard_router.bucket_id_mpcrc32(identifier.hash)

    local ids, err = M.vshard_router.callrw(bucket_id, 'search_storage_api.user_id.get_by_phone_number',
        {identifier.hash, identifier.data}, {timeout = M.vshard_timeout}
    )
    if err ~= nil then
        err = errors.wrap(err)
        return nil, err
    end
    return ids
end

-- search_storage_api.lua
local M = {}
local function get_users_by_hash(hash, data, cmp_func)
    local result = {}
    for _, t in box.space.user_search_index.index.hash:pairs({hash}, 'EQ') do
        if t.data_hash ~= hash then
            break
        end
        if cmp_func(t.data, data) then
            result[#result + 1] = t.user_id
        end
    end
    if #result == 0 then
        return nil, not_found_err:new()
    end
    return result
end

function M.user_id.get_by_phone_number(hash, data)
    return get_users_by_hash(hash, data, M.cmp_data.phone_number)
end

Обошлись без MapReduce. В случае уникальных данных делаем один сетевой запрос для нахождения primary_key, и второй для получения искомых данных.

Индексация произвольных данных

Индексный слой позволяет перенести индексацию с уровня БД на уровень кода приложения. Это открывает широкие возможности по индексации произвольных данных. Например, можно достаточно легко сделать составной индекс на основе имени и даты рождения.

-- identifier.lua

local M = {}

local digest = require('digest')
local msgpack = require('msgpack')
M.cmp_data = {}

local Identifier = {}

function Identifier:new(data)
    local obj = {}
    obj.hash = digest.md5_hex(msgpack.encode(data))
    obj.data = data
    return obj
end

...

function M.name_birthdate(name, birthdate)
    local data = {'name', name, 'birthdate', birthdate}
    return Identifier:new(data)
end

function M.cmp_data.name_birthdate(data_one, data_two)
    return #data_one == 4 and #data_two == 4 and
            (data_one[1] == data_two[1]) and
            (data_one[2] == data_two[2] ~= nil) and
            (data_one[3] == data_two[3] ~= nil) and
            (data_one[4] == data_two[4] ~= nil)
end

return M

Остальное полностью аналогично поиску по номеру телефона.

Write

Выше я описал, как искать с помощью индексного слоя. Теперь рассмотрим его построение. При создании или обновлении данных необходимо создавать или обновлять и индексный слой. Реализация этих процессов зависит от конкретной задачи, но скорее всего она будет идти в фоне. Для этого можно воспользоваться модулями Tarantool/queue и Tarantool/sharded-queue. В этом примере есть наивная реализация построения индексного слоя с помощью Tarantool/queue.

При реализации индексного слоя мы усложним запись и обновление данных, зато существенно ускорим чтение. В нашем случае работа с индексным слоем при записи выглядит так:

Обновление выполняется в фоне уже после того, как данные обновились. То есть  возможна ситуация, когда поиск ничего не вернёт, но данные в БД уже будут. Или в случае обновления поиск по старым данным вернёт новые, так как содержимое обновилось, а поисковый слой ещё нет. Но в конечном итоге данные будут согласованы. Возможно, в вашем случае нужна иная логика: сначала обновить индексный слой, а затем — данные.

Итоги

Индексный слой: 

• Позволит существенно уменьшить количество сетевых запросов на чтение при поиске уникальных данных.

• Позволит не хранить производные данные для поиска по ним.

• Позволит реализовать быстрый поиск по любым уникальным данным, хранящимся у вас.

Скачать Tarantool можно на официальном сайте, а получить помощь — в Telegram-чате.