0. Введение
В этой статье описывается опыт Didi по построению NewSQL поверх распределенного хранилища Nosql Fusion. Подробно опишите функции, сценарии приложений и дизайнерские решения Fusion-NewSQL.
1. Предпосылки
Fusion-NewSQL — это система хранения NewSQL, разработанная Didi и построенная на основе распределенного хранилища KV. Fusion-NewSQ совместим с протоколом MySQL, поддерживает функцию вторичного индекса и обеспечивает сверхбольшое постоянное хранилище данных и высокопроизводительное чтение и запись.
▍Наши вопросы
Быстрое и непрерывное развитие бизнеса Didi, быстрый рост объема данных и запросов, а также возрастающая нагрузка на системы хранения. Хотя подбаза данных и подтаблица могут в определенной степени удовлетворить потребности в увеличении объема данных и запросов, из-за быстрых изменений в бизнесе нескольких бизнес-направлений DiDi (экспресс, специальный автомобиль, двухколесный транспорт и т. , потребность в добавлении полей и индексов в базу данных возникает очень часто, схема подтаблиц подбазы данных не подходит для частых операций изменения схемы, что приведет к тяжелым задачам администратора баз данных и длительным циклам изменений, а также будет иметь определенную влияние на онлайн-операции операций с огромными таблицами. В то же время схема подтаблиц подбазы данных не дружелюбна к поддержке вторичного индекса или вообще не поддерживает ее. Ввиду вышеописанной ситуации решение базы данных NewSQL стало для нас направлением решения бизнес-задач.
▍Исследование продуктов с открытым исходным кодом
Вначале мы исследовали распределенное решение NewSQL с открытым исходным кодом: TIDB. Хотя TIDB — очень хороший продукт NewSQL, для нашего бизнес-сценария TIDB не очень подходит по следующим причинам:
Нам нужно решение для базы данных с высокой пропускной способностью и малой задержкой, но поскольку TIDB необходимо выполнять транзакции, решение из 2 компьютеров, естественно, не может обеспечить низкую задержку (99rt в течение 100 мс или даже 99rt в течение 50 мс). Большинству наших предприятий на самом деле не нужны распределенные транзакции, или мы можем обойти распределенные транзакции с помощью других механизмов компенсации. Это связано с бизнес-сценарием. Стоимость места для хранения трех копий TIDB относительно высока. Некоторые из наших внутренних автономных данных импортируются в онлайн-систему, которая не может быть напрямую связана с TIDB.
Основываясь на вышеуказанных причинах, мы начали путь самостоятельной разработки NewSQL, отвечающей потребностям нашего бизнеса.
▍Наш фонд
Мы не собираемся разрабатывать полноценную систему NewSQL с нуля, а строим NewSQL, который может соответствовать нашим бизнес-сценариям, на базе распределенного хранилища KV Fusion собственной разработки. Fusion — это база данных NoSQL, использующая архитектуру Codis, совместимая с протоколом и структурой данных Redis и использующая Rocksdb в качестве механизма хранения. Fusion используется сотнями компаний в Didi, и это одно из основных онлайн-хранилищ Didi.
Схема архитектуры Fusion выглядит следующим образом:
Мы используем хэш-шардинг для разделения данных. Сверху вниз пользователи могут получить доступ к Fusion через клиент протокола Redis.Запрос пользователя на доступ отправляется на прокси-сервер, а прокси-сервер пересылает данные на узел данных серверной части Fusion. Пересылка прокси-сервера на серверный узел данных заключается в вычислении хеш-значения в соответствии с запрошенным ключом, а затем использовании остатка количества сегментов слота для получения фиксированного идентификатора слота.Каждый слот будет фиксированно сопоставлен с хранилищем. node для решения проблемы маршрутизации данных.
Благодаря высокому уровню параллелизма, малой задержке и большой емкости хранилища все, что нам нужно сделать, — это построить протокол MySQL и вторичные индексы поверх него. Итак, как преобразовать формат данных MySQL в хранилище структуры данных Redis — это проблема, с которой мы должны столкнуться и которая будет подробно описана позже.
2. Спрос
Высокая пропускная способность, низкая задержка и большая емкость. Совместимость с протоколом MySQL и экологией нижестоящего уровня. Поддержка запроса первичного ключа и запроса вторичного индекса. Изменения схемы являются гибкими и не влияют на стабильность онлайн-сервисов.
3. Архитектурный дизайн
Fusion-NewSQL состоит из следующих частей:
DiseServer анализирует протокол MySQL Кластер Fusion для хранения данных - Кластер данных Кластер Fusion для хранения информации об индексах — Кластер индексов Центр конфигурации управления, отвечающий за схему — ConfigServer Программа построения асинхронного индекса-Потребитель отвечает за использование данных формата MySQL-Binlog, записанных кластером данных в MQ, а также за создание данных индекса и запись их в кластер индекса в соответствии с информацией схемы. Внешние зависимости, MQ, Zookeeper
Схема архитектуры выглядит следующим образом:
4. Детальный дизайн
▍Структура хранения
Как преобразовать данные структуры таблицы MySQL в структуру данных Redis — это первая проблема, с которой мы сталкиваемся.
Как показано ниже:
Мы преобразуем строку записей в таблице MySQL в структуру Hashmap Redis. Ключ Hashmap состоит из имени таблицы и значения первичного ключа, что соответствует глобальной уникальной функции. На следующем рисунке показано, как MySQL преобразуется в протокол Redis с помощью запроса первичного ключа:
В дополнение к данным, в Fusion-NewSQL также необходимо хранить индексы.В отличие от данных, хранящихся в виде хэш-карт, индексы хранятся в виде структур ключ-значение. В зависимости от типа индекса существует небольшая разница в формате ключ-значение (приведенный ниже формат на самом деле закодирован для разделителя и имени индекса, чтобы он выглядел интуитивно понятным):
Уникальный индекс: Ключ: table_indexname_indexColumnsValue Значение: Роки
Неуникальный индекс: Ключ: table_indexname_indexColumnsValue_Rowkey Значение: ноль
Причина такой разницы в том, что часть неуникального индекса перед добавлением Rowkey может повторяться и не может быть глобально уникальной. Кроме того, уникальный индекс не кодирует Rowkey в ключе, потому что, когда оператор запроса представляет собой простой запрос «=», соответствующее содержимое Rowkey можно найти с помощью операции прямого получения без сканирования, что более эффективно.
Позже, в процессе запроса, мы сосредоточимся на том, как запрашивать данные через вторичные индексы.
▍Процесс чтения и записи данных
запись данных
Пользователь отправляет протокол на dise-сервер через MySQL-sdk. dise-server проверяет SQL, написанный пользователем, в соответствии со схемой Dise-сервер преобразует SQL, прошедший проверку, в структуру Hashmap Redis и отправляет его в кластер данных по протоколу Redis. Кластер данных записывает данные в файл wal и сохраняет данные в rockdb. Фоновый поток кластера данных использует wal-файл и преобразует его в формат MySQL-Binlog. Отправить данные в MQ Модуль асинхронного индекса использует MQ, объединяет MySQL-Binlog с информацией о схеме в соответствии с типом операции (вставка, обновление, удаление), создает информацию индекса и записывает данные индекса в кластер индекса. По указанной выше ссылке пользовательская операция записи MySQL завершает хранение данных и построение индекса. Поскольку шаг построения индекса по данным выполняется асинхронно через MQ, между данными и индексом будет определенная разница во времени.
Запрос
Ниже приведен пример запроса данных с использованием вторичного индекса:
Dise-сервер получает SQL-запрос, выбирает индекс в соответствии с условиями и возвращает пользователю ошибку, если индекс не найден (Fusion-NewSQL не может использовать неиндексные поля в качестве условий запроса). В соответствии с выбранным индексом создайте диапазон запроса и используйте команду сканирования для обхода кластера индексов, чтобы получить набор первичных ключей, соответствующих условиям. На следующем рисунке показан пример использования сканирования для обхода вторичного индекса с помощью SQL-запроса:
В соответствии с первичным ключом используйте команду hgetall, чтобы запросить у кластера данных набор результатов, соответствующий условиям. Встройте результирующий набор в результаты MySQL и верните их пользователю. В соответствии с форматом данных индекса выше, при сканировании диапазона префикс должен быть фиксированным, а когда он отображается в оператор SQL, это означает, что в условии where to запрос диапазона может иметь только одно поле, не несколько полей. Например:
Индекс представляет собой совместный индекс полей возраста и имени. Если оператор запроса выглядит следующим образом:
выберите * у учащегося, у которого возраст > 20 и имя > 'W';
Сканирование не может определить префикс, и невозможно запросить данные, соответствующие условиям, через индекс index_age_name, поэтому использование формы KV для сохранения в индексе может удовлетворить только условию where. является запросом диапазона. Конечно, это можно решить, храня объединенный индекс отдельно и ища пересечение несколько раз, но это противоречит нашему первоначальному намерению уменьшить количество RPC и уменьшить задержку. Чтобы решить эту проблему, мы представили поисковую систему Elastic Search, которая будет подробно описана далее в этом разделе.
▍Изменение схемы
Когда пользователь вносит изменения в схему, они будут отправлены в систему управления в виде рабочего задания. После одобрения системой управления и контроля запрос на изменение будет отправлен в центр конфигурации.После того, как центр конфигурации выполнит проверку безопасности, новая схема будет записана в хранилище, и изменения будут отправлены на каждый узел.
Изменения поля:
Узел получает push-уведомление и обновляет локальную схему. Для исторических данных он на самом деле не изменяет данные, но при запросе сопоставляет поля в соответствии с информацией схемы.Если в данных отсутствуют некоторые поля, чем в схеме, вместо этого используйте значение по умолчанию; если в данных больше полей, чем в схеме , оно скрыто, лишние поля не отображаются.
Добавление индекса делится на два этапа:
Добавляются новые индексы, исторические данные не обрабатываются, а добавочные данные сразу же проходят через процесс построения индекса. Используйте инструмент построения исторического индекса, чтобы просмотреть исторические данные, построить KV нового индекса и завершить построение индекса на основе исторических данных. Здесь есть пункт оптимизации, сканируйте ведомое устройство вместо ведущего, чтобы не влиять на линию.
5. Экологическое строительство
Эпоха единого продукта для решения всех проблем давно прошла. Хранилища данных не могут хорошо служить бизнесу. Fusion-NewSQL был разработан для подключения к другим системам хранения со дня его разработки.
▍Fusion-NewSQL для других систем хранения
Fusion-NewSQL отправляет данные в MQ в формате Binlog, совместимом с MySQL. Все подчиненные системы, которые могут получить доступ к данным MySQL, могут хранить данные в других системах, используя данные Fusion-NewSQL в том же формате в MQ. Такой подход обеспечивает максимальную совместимость с наименьшим объемом работы.
▍Переход на Fusion-NewSQL
Fusion-NewSQL также поддерживает перенос данных из автономных таблиц Hive в Fusion-NewSQL с помощью инструмента FastLoad (DTS), предоставляемого Fusion-NewSQL, который удовлетворяет потоку автономных данных в онлайн-данные.
Если пользователь завершает поток данных самостоятельно, таблица Hive обычно сканируется, затем создается оператор записи MySQL, и данные записываются в Fusion-NewSQL один за другим Процесс выглядит следующим образом:
MySQL-клиент отправляет запросы на запись в DiseServer. DiseServer записывает MySQL для синтаксического анализа, преобразует его в хэш-карту и отправляет преобразованные данные в кластер данных с использованием протокола Redis. Узел хранения кластера данных получает данные и записывает их в файл wal. Узлы хранения кластера данных следуют процессу записи Rocksdb, который включает в себя запись memtable, также возможно, что memtable заполнена, происходит сброс и срабатывает фоновое уплотнение. Асинхронные потоки потребляют wal и отправляют данные в MQ в формате MySQL-Binlog. Программа асинхронного индексирования использует MySQL-Binlog, создает данные, необходимые кластеру Index, и отправляет запросы на запись в кластер Index. Узел хранения кластера Index записывает wal. Узлы хранения кластера Index входят в процесс записи Rocksdb.
Из приведенного выше процесса мы видим, что у этого метода миграции есть несколько болевых точек:
Пользователям, у которых есть такие требования к импорту данных из Hive в Fusion-NewSQL, необходимо разработать набор кода с той же логикой, а стоимость обслуживания высока. Каждые данные Hive должны проходить через длинную ссылку, а импорт данных занимает много времени. Оффлайн-платформа имеет большой объем данных и высокую пропускную способность, что напрямую увеличивает QPS онлайн-системы и оказывает большое влияние на стабильность онлайн-системы.
Основываясь на вышеупомянутых болевых точках, мы разработали платформу импорта данных Fastload, согласовав формат таблицы с Hive на Fusion-NewSQL, используя Hadoop для параллельной обработки данных и создав файлы хранилища sst, которые может распознать Rocksdb, минуя сложную ссылку записи DISE. , напрямую Для импорта данных в Fusion-NewSQL процесс выглядит следующим образом:
Пользователь заполняет заказ на работу и выбирает сопоставление некоторых полей указанной таблицы Hive с полями таблицы Fusion-NewSQL (здесь несколько полей в Hive могут образовывать поле Fusion-NewSQL). Hadoop просматривает таблицу Hive и получает информацию о маршрутизации, согласно которой данные должны храниться в кластере данных и кластере индексов, через Zookeeper. Благодаря описанному выше обходу, вычислению, а затем данные напрямую преобразуются в sst, который Rocksdb может распознать, а данные, хранящиеся в нем, уже представляют собой данные KV, состоящие из информации о структуре таблицы DISE. Отправьте sst-файл непосредственно на указанный узел хранения.Узел хранения или функция загрузки, предоставляемая Rocksdb, напрямую загружает sst-файл в Fusion-NewSQL, и пользователи могут его прочитать.
Это решение позволяет избежать длинных и сложных ссылок на запись, не увеличивает QPS системы, практически не влияет на онлайн-доступ, если дисковый и сетевой ввод-вывод не достигает узкого места, при этом пользователям нужно только заполнить Отношения сопоставления Hive to Fusion-NewSQL Schema достаточно, и вам не нужно заботиться о реализации.
▍Реализовать сложные запросы с помощью Elastic Search
В процессе использования MySQL или Fusion-NewSQL для бизнеса мы обнаружили, что существует такой сценарий: условия запроса бизнеса очень сложны, а количество задействованных полей, условий и агрегаций относительно велико. сценарий, компания выберет использование Elastic Search в качестве нижестоящего последовательности MySQL или Fusion-NewSQL, импортирует данные в Elastic Search, а затем использует богатые возможности поиска Elastic Search, чтобы сначала получить первичный ключ данных в MySQL или Fusion-NewSQL из Elastic. Выполните поиск, а затем получите все данные на основе первичного ключа.
В соответствии с приведенным выше сценарием Fusion-NewSQL предоставляет специальный тип индекса: ES. Когда пользователи создают индекс, они могут выбрать поля, которые должны выполнять сложные запросы, и вместе построить индекс ES, который не только отвечает потребностям бизнеса, но и позволяет избежать необходимости разрабатывать сложную логику для взаимодействия с Elastic Search для каждого бизнеса. и унифицированный интерфейс базы данных для MySQL. В то же время это также компенсирует упомянутую выше способность вторичного индекса KV Fusion-NewSQL не поддерживать извлечение нескольких диапазонов полей.
Схема архитектуры выглядит следующим образом:
Индекс ES отмечен красной цифрой 4 на рисунке выше, а информация о поле и первичный ключ, содержащиеся в индексе ES, записываются в Elastic Search. При запросе зеленого цвета 1, если выбран индекс типа ES, он соберет оператор DSL Elastic Search в соответствии с полями, участвующими в условии where, получит первичный ключ из Elastic Search, а затем получит его из кластера данных. Из-за медленной задержки запросов Elastic Search Fusion-NewSQL может поддерживать несколько индексов в таблице, используя индексы KV и индексы ES для сосуществования.Для требований с высокой задержкой условия запроса относительно просты для использования индексов KV; для сложных условий запроса, Требования к задержке Не высокие, используйте индекс ES.
6. Резюме
Fusion-NewSQL подключен к 70 основным подразделениям, таким как заказы, оценки, счета, пользовательские центры и механизмы транзакций.Общее количество запросов в секунду превышает 200 Вт, а общий объем данных превышает 600 ТБ.
Конечно, Fusion-New — это не общее и полное решение NewSQL, а на основе существующей базы данных nosql, за счет поддержки протокола SQL и комбинации различных компонентов построить базу данных для внешнего выражения, но в этом Таким образом, это может быть С наименьшими затратами на разработку, он может соответствовать большинству бизнес-сценариев и иметь высокий коэффициент ввода-вывода.
7. Последующая работа
Ограниченная поддержка транзакций, например, данные, которые позволяют бизнес-планированию ложиться на один узел, могут поддерживать межсетевые транзакции на одной машине. Индексирование в реальном времени заменяет асинхронное индексирование, которое удовлетворяет требованиям как для записи, так и для чтения. В настоящее время существует схема механизма сквозной записи + компенсации, которая может удовлетворить индекс в реальном времени в нормальном состоянии без распределенных транзакций и обеспечить окончательную согласованность индекса данных в ненормальных условиях. Поддержка дополнительных протоколов и функций SQL.
Автор этой статьи: ▬
Ли Синь Диди | Старший инженер по разработке программного обеспечения
Многолетний опыт проектирования и разработки в области распределенного хранения. Он участвовал в проектировании и разработке базы данных Nosql/NewSQL Fusion, распределенной базы данных временных рядов sentry, базы данных NewSQL SDB и других систем.