О загрузке данных
Aug. 29th, 2014 03:35 pm![[personal profile]](https://www.dreamwidth.org/img/silk/identity/user.png){kind=small}
kaipaРедкая запись на тему моей профессиональной деятельности :)
Вот тут один товарищ пишет, как в Фейсбуке (не соц.сети, а компании) с его, видимо, помощью сделали систему, загружающую в СУБД Вертика 35TB/час данных (или 10GB/сек). Это нижняя планка, SLA, а пиковая производительность должна быть в полтора-два раза выше, чтобы реагировать на колебания нагрузки, незначительные задержки и прочее. Необходимо заметить, что параллельно непрерывной загрузке система обслуживает запросы клиентов. В нашей компании мы загружаем данных меньше, но в расчете на узел кластера, думаю, примерно столько же. Я давно хочу написать статью на Хабре про оптимизацию скорости загрузки в Вертику, но чтобы не быть голословным, надо заново перепрогнать тесты в разных конфигурациях, нарисовать таблички и графики, а это время и лень. Тем не менее, базовые принципы, изложенные в статье выше, верные и могут быть применены к любой массивно-параллельной системе управления данными, не только к Вертике. Попробую их тут обобщить, по ходу дела комментируя некоторые моменты вышеуказанной статьи.
Для начала, некоторые предварительные соображения. Далее я буду употреблять слово файл для обозначения порции данных (обычно довольно большой). Можно заливать напрямую потоком через сеть, но такое решение не очень устойчиво к отказам сети или заливающего узла кластера. Кроме того, я буду употреблять слово сервер для обозначения узла кластера, мне кажется это более естественным. И последнее, я буду писать диск понимая под этим дисковую подсистему.
Скорость загрузки или правильнее говорить пропускная способность системы на запись, так как загрузка происходит во много потоков, определяется сочетанием трех основных факторов:
1. Скоростью обработки и загрузки одного файла на одном сервере.
2. Масштабируемостью по числу параллельных загрузок в пределах одного сервера.
3. Масштабируемостью по числу параллельных загрузок в пределах всего кластера.
Сверху скорость, очевидно, ограничена производительностью дисковой подсистемы.
Пропускной способностью сети можно пренебречь. Гигабитовая сеть (желательно на выделеном свитче) работает быстрее диска.
Отступление на тему дисков.
Сейчас активно идет наступление твердотельных носителей (SSD). Но это наступление идет в основном на фронте восокопроизводительных транзакционных систем (OLTP), где критически важно короткое время поиска на диске (random seek). Для аналитических систем больших данных (OLAP) преимущества SSD не столь существенны (если вообще есть), так как пишутся и читаются не отдельные записи, а большие блоки данных. Разница в цене не оправдывается. Поэтому целесообразно строить дисковую подсистему как RAID 10 на восьми, скажем, не очень больших быстрых дисках (500-1000GB, 15Krpm SAS), что дает, за вычетом системных файлов, 1.7-3.5TB на сервер. SSD все же может иметь применение, но как небольшой массив для временных файлов, например (примерно так построены большие системы у TeraData).
Итак, вернемся к загрузке. Может показаться, что существует идеальное решение: каждый файл грузится только на свой выделенный сервер кластера. Очевидно, что при достаточно большом числе файлов можно 100% занять каждый сервер, а масштабируемость по числу серверов линейная. Основная проблема в таком решении -- надежность. Отказ сервера означает потерю сегмента данных. Во всех распределенных системах, где данные чего-то стоят, каждый сегмент данных хранится как минимум в двух копиях на разных серверах. Поэтому идеальное решение работает только в том случае, где каждый сервер продублирован. Но это тоже не хорошо, так как во-первых, идеальное решение сразу перестает быть идеальным, а во-вторых, в случае отказа сервера и его последующей замены на резервный сервер падает двойная нагрузка: он не только должен продолжать грузить данные, но и отдавать их для восстановления замененного. Поэтому обычно данные грузятся равномерно на все или почти все сервера кластера, обеспечивая равномерную нагрузку на процессоры и диски.
Теперь, как это делается в Вертике. Загрузка файла выполняется посредством одного SQL-оператора COPY, но внутри состоит из следующих связанных между собой этапов:
1. Чтение файла с диска
2. Распаковка, если файл запакован.
3. Синтаксический разбор (парсинг)
4. Преобразование данных в структуры базы данных и сортировка
5. Запись на диск
Пункты 1-2 локальны к серверу загрузки, 3 -- может быть как локальным так и распределенным, 4-5 --- распределенные по всем серверам. Разработчики Вертики очень сильно оптимизировали пункты 4-5, и здесь мало что можно улучшить, разве что аккуратно подходить к физическому дизайну, так как это непосредственно влияет на скорость (4) и размер данных, которые пишутся в пункте 5.
Основные эксперименты можно производить в пределах 1-3. И тут есть следующие возможности:
- файл распакованный и запакованный
- файл в CSV формате или бинарный (Vertica native)
- файл грузится/распоковывается Вертикой, или внешней утилитой и скрамливается Вертике через поток
- файл грузится локально на сервере или распределенно на нескольких или всех серверах
- сколько одновременно файлов можно грузить на одном сервере
- сколько одновременно файлов можно грузить на кластере в целом
- оптимальный размер файла
- и т.д.
Наилучшее решение зависит от размера файлов и конкретной конфигурации сервера и кластера, но после наших экспериментов мы остановились на следующем: файл gzip в бинарном формате грузится Вертикой параллельно на нескольких (не всех) серверах.
В пределах одного сервера Вертика распараллеливает распаковку и парсинг на все ядра, обеспечивая 100% загрузку процессора. Для того, чтобы распараллелить на несколько серверов, в версиях Вертики до 7.0 была отдельная библиотека PLoad, но она обладала сильными ограничениями. Позднее, база данных вроде бы стала сама распараллеливать COPY, но мы этого почему-то не заметили. Другой официальный способ состоит в использовании общей файловой системы, тогда Вертику можно проиструктировать читать файл одновременно всеми серверами и распределять работу между ними автоматически. Или же можно один и тот же файл покладывать для загрузки на все сервера, но это означает многократное увеличение сетевого трафика. Если эти варианты не удобны, то есть слабо документированный способ параллельной обработки, который работает во всех версиях. Для этого необходимо файл делить на несколько частей заранее, и указывать в COPY несколько файлов сразу (или маску) -- в этом случае разные части файлов Вертика посылает на обработку на разные сервера, хотя читаются они с одного.
Оптимальное количество частей файлов и параллельных потоков загрузки определяется экспериментальным образом. У нас получалось, что количество параллельных загрузок должно быть примерно в два раза меньше количества серверов. Но учитывая, что каждый поток разбивался на несколько частей, общее число потоков загрузки в 2-3 раза больше количества серверов. При увеличении количества потоков или частей скорость загрузки все еще растет, но сильно падает скорость запросов и агрегации данных.
Похожая ситуация описана и в статье, и там предложен еще один интересный способ оптимизации -- эфемерные (Ephemeral) сервера. Это сервера в составе кластера, на которых нет данных. Обычно они "получаются" на стадии расширения кластера: сначала добавляют новые сервера, потом перераспределяют данные. Но их можно оставить и пустыми и использовать немного нестандартно. В этом случае появляется возможность делать загрузку и парсинг только на пустых эфемерных серверах, не нагружая остальные сервера с данными. Основное преимущество в этом случае не в скорости загрузки на чистой системе, а в том, что клиенты не "дерутся" за процессор с загрузкой, так как все обработка клиентских запросов происходит на серверах с данными. Дополнительное преимущество в том, что чтение файла не конфликтует с записью данных, если для файлов и базы данных используется один и тот же диск.
Еще пара замечаний по статье.
Автор пишет, что в фейсбуковском 3.5-4PB (петабайт) кластере каждые три дня обновляется примерно 1.5PB. То есть большинство данных коротко живущие. Я точно не считал, но у нас соотношение примерно похожее.
Кроме того, он указывает, что стандартное решение для надежности -- это иметь две параллельные идентичные системы, в которые одинаковые данные загружаются независимо (The most common practice for a standby cluster is to run an ETL twice to maintain two separate instances of Vertica). Когда я несколько лет назад написал об этом на Хабре (мы до этой идеи дошли совершенно независимо), меня заминусовали.
Вот такой набросок. Статью на Хабре все же напишу.
Вот тут один товарищ пишет, как в Фейсбуке (не соц.сети, а компании) с его, видимо, помощью сделали систему, загружающую в СУБД Вертика 35TB/час данных (или 10GB/сек). Это нижняя планка, SLA, а пиковая производительность должна быть в полтора-два раза выше, чтобы реагировать на колебания нагрузки, незначительные задержки и прочее. Необходимо заметить, что параллельно непрерывной загрузке система обслуживает запросы клиентов. В нашей компании мы загружаем данных меньше, но в расчете на узел кластера, думаю, примерно столько же. Я давно хочу написать статью на Хабре про оптимизацию скорости загрузки в Вертику, но чтобы не быть голословным, надо заново перепрогнать тесты в разных конфигурациях, нарисовать таблички и графики, а это время и лень. Тем не менее, базовые принципы, изложенные в статье выше, верные и могут быть применены к любой массивно-параллельной системе управления данными, не только к Вертике. Попробую их тут обобщить, по ходу дела комментируя некоторые моменты вышеуказанной статьи.
Для начала, некоторые предварительные соображения. Далее я буду употреблять слово файл для обозначения порции данных (обычно довольно большой). Можно заливать напрямую потоком через сеть, но такое решение не очень устойчиво к отказам сети или заливающего узла кластера. Кроме того, я буду употреблять слово сервер для обозначения узла кластера, мне кажется это более естественным. И последнее, я буду писать диск понимая под этим дисковую подсистему.
Скорость загрузки или правильнее говорить пропускная способность системы на запись, так как загрузка происходит во много потоков, определяется сочетанием трех основных факторов:
1. Скоростью обработки и загрузки одного файла на одном сервере.
2. Масштабируемостью по числу параллельных загрузок в пределах одного сервера.
3. Масштабируемостью по числу параллельных загрузок в пределах всего кластера.
Сверху скорость, очевидно, ограничена производительностью дисковой подсистемы.
Пропускной способностью сети можно пренебречь. Гигабитовая сеть (желательно на выделеном свитче) работает быстрее диска.
Отступление на тему дисков.
Сейчас активно идет наступление твердотельных носителей (SSD). Но это наступление идет в основном на фронте восокопроизводительных транзакционных систем (OLTP), где критически важно короткое время поиска на диске (random seek). Для аналитических систем больших данных (OLAP) преимущества SSD не столь существенны (если вообще есть), так как пишутся и читаются не отдельные записи, а большие блоки данных. Разница в цене не оправдывается. Поэтому целесообразно строить дисковую подсистему как RAID 10 на восьми, скажем, не очень больших быстрых дисках (500-1000GB, 15Krpm SAS), что дает, за вычетом системных файлов, 1.7-3.5TB на сервер. SSD все же может иметь применение, но как небольшой массив для временных файлов, например (примерно так построены большие системы у TeraData).
Итак, вернемся к загрузке. Может показаться, что существует идеальное решение: каждый файл грузится только на свой выделенный сервер кластера. Очевидно, что при достаточно большом числе файлов можно 100% занять каждый сервер, а масштабируемость по числу серверов линейная. Основная проблема в таком решении -- надежность. Отказ сервера означает потерю сегмента данных. Во всех распределенных системах, где данные чего-то стоят, каждый сегмент данных хранится как минимум в двух копиях на разных серверах. Поэтому идеальное решение работает только в том случае, где каждый сервер продублирован. Но это тоже не хорошо, так как во-первых, идеальное решение сразу перестает быть идеальным, а во-вторых, в случае отказа сервера и его последующей замены на резервный сервер падает двойная нагрузка: он не только должен продолжать грузить данные, но и отдавать их для восстановления замененного. Поэтому обычно данные грузятся равномерно на все или почти все сервера кластера, обеспечивая равномерную нагрузку на процессоры и диски.
Теперь, как это делается в Вертике. Загрузка файла выполняется посредством одного SQL-оператора COPY, но внутри состоит из следующих связанных между собой этапов:
1. Чтение файла с диска
2. Распаковка, если файл запакован.
3. Синтаксический разбор (парсинг)
4. Преобразование данных в структуры базы данных и сортировка
5. Запись на диск
Пункты 1-2 локальны к серверу загрузки, 3 -- может быть как локальным так и распределенным, 4-5 --- распределенные по всем серверам. Разработчики Вертики очень сильно оптимизировали пункты 4-5, и здесь мало что можно улучшить, разве что аккуратно подходить к физическому дизайну, так как это непосредственно влияет на скорость (4) и размер данных, которые пишутся в пункте 5.
Основные эксперименты можно производить в пределах 1-3. И тут есть следующие возможности:
- файл распакованный и запакованный
- файл в CSV формате или бинарный (Vertica native)
- файл грузится/распоковывается Вертикой, или внешней утилитой и скрамливается Вертике через поток
- файл грузится локально на сервере или распределенно на нескольких или всех серверах
- сколько одновременно файлов можно грузить на одном сервере
- сколько одновременно файлов можно грузить на кластере в целом
- оптимальный размер файла
- и т.д.
Наилучшее решение зависит от размера файлов и конкретной конфигурации сервера и кластера, но после наших экспериментов мы остановились на следующем: файл gzip в бинарном формате грузится Вертикой параллельно на нескольких (не всех) серверах.
В пределах одного сервера Вертика распараллеливает распаковку и парсинг на все ядра, обеспечивая 100% загрузку процессора. Для того, чтобы распараллелить на несколько серверов, в версиях Вертики до 7.0 была отдельная библиотека PLoad, но она обладала сильными ограничениями. Позднее, база данных вроде бы стала сама распараллеливать COPY, но мы этого почему-то не заметили. Другой официальный способ состоит в использовании общей файловой системы, тогда Вертику можно проиструктировать читать файл одновременно всеми серверами и распределять работу между ними автоматически. Или же можно один и тот же файл покладывать для загрузки на все сервера, но это означает многократное увеличение сетевого трафика. Если эти варианты не удобны, то есть слабо документированный способ параллельной обработки, который работает во всех версиях. Для этого необходимо файл делить на несколько частей заранее, и указывать в COPY несколько файлов сразу (или маску) -- в этом случае разные части файлов Вертика посылает на обработку на разные сервера, хотя читаются они с одного.
Оптимальное количество частей файлов и параллельных потоков загрузки определяется экспериментальным образом. У нас получалось, что количество параллельных загрузок должно быть примерно в два раза меньше количества серверов. Но учитывая, что каждый поток разбивался на несколько частей, общее число потоков загрузки в 2-3 раза больше количества серверов. При увеличении количества потоков или частей скорость загрузки все еще растет, но сильно падает скорость запросов и агрегации данных.
Похожая ситуация описана и в статье, и там предложен еще один интересный способ оптимизации -- эфемерные (Ephemeral) сервера. Это сервера в составе кластера, на которых нет данных. Обычно они "получаются" на стадии расширения кластера: сначала добавляют новые сервера, потом перераспределяют данные. Но их можно оставить и пустыми и использовать немного нестандартно. В этом случае появляется возможность делать загрузку и парсинг только на пустых эфемерных серверах, не нагружая остальные сервера с данными. Основное преимущество в этом случае не в скорости загрузки на чистой системе, а в том, что клиенты не "дерутся" за процессор с загрузкой, так как все обработка клиентских запросов происходит на серверах с данными. Дополнительное преимущество в том, что чтение файла не конфликтует с записью данных, если для файлов и базы данных используется один и тот же диск.
Еще пара замечаний по статье.
Автор пишет, что в фейсбуковском 3.5-4PB (петабайт) кластере каждые три дня обновляется примерно 1.5PB. То есть большинство данных коротко живущие. Я точно не считал, но у нас соотношение примерно похожее.
Кроме того, он указывает, что стандартное решение для надежности -- это иметь две параллельные идентичные системы, в которые одинаковые данные загружаются независимо (The most common practice for a standby cluster is to run an ETL twice to maintain two separate instances of Vertica). Когда я несколько лет назад написал об этом на Хабре (мы до этой идеи дошли совершенно независимо), меня заминусовали.
Вот такой набросок. Статью на Хабре все же напишу.
![[identity profile]](https://www.dreamwidth.org/img/silk/identity/openid.png){kind=small}
no subject
Date: 2014-09-02 12:22 pm (UTC)