Обзор надежности систем загрузки журнальных записей в Big Data

Данная статья фокусируется на обзоре методов надежности ныне существующих систем для загрузки журнальных записей и их реализации. Дано определение надежности таких систем. Определены методы загрузки. Освещена особенность систем потоковой обработки в виде гарантии доставки сообщений. По каждому из блоков выделены результаты обзора каждой из систем.

Ключевые слова: Big Data, Apache Software Foundation, Kafka, Flume, Flink, Sqoop, Chukwa, Gobblin, Storm, Apex, NiFi, Fluentd, надежность систем загрузки, пакетная загрузка, потоковая загрузка, гарантия доставки сообщений.

Использование Big data систем анализа данных невозможно без использования системы загрузки данных для их последующей обработки. При использовании больших данных возникает необходимость надежной загрузки этих данных и того как системы загрузки справляются с этой задачей.

Определение надежности

Надежность систем загрузки журнальных записей определяется отказоустойчивостью при загрузке или выгрузке данных, наличием механизмов предотвращения ошибок и возможностью восстановления передачи данных при сбое [1].

Классификация методов загрузки

В Big Data существуют несколько видов загрузки данных:

– Пакетная загрузка — Данный вид используется для очень больших файлов или, когда быстрое время отклика не критично. Файлы, которые нужно передать, собираются в течение определенного периода времени, а затем отправляются вместе в виде пакетов.

– Потоковая загрузка — Данный вид используется если данные поступают в реальном времени и должны быть загружены в систему анализа данных незамедлительно.

Надежность систем пакетной обработки

В качестве систем пакетной обработки обозреваются Chukwa, Gobblin и Sqoop. Так же следует отметить, что системы Flink и Apex Malhar также подлежат рассмотрению, поскольку имеют в своем арсенале возможность работать как в режиме потоковой обработки, так и в режиме пакетной.

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

Так, например, в Chukwa коллекторы записывают данные в HDFS, и как только HDFS в качестве подтверждения записи возвращает успех, коллекторы сообщают об этом агенту, который запоминает состояние контрольной точки [5].

В Apex же это решается путем записи того, сколько оператор записал в каждый файл. При повторной передаче данное состояние проверяется и в случае не состыковки оператор усекает файлы обратно на контрольную точку восстановления [6].

Вторым по распространенности подходом является повтор отправления данных.

При неудачной попытке отправления Gobblin и Flink повторяют отправку данных настраиваемое количество раз, при этом в Flink помимо прочего имеется возможность задержки отправки при следующем повторе [7].

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

HDFS обеспечивает механизм отказоустойчивости посредством процесса репликации. В HDFS всякий раз, когда файл сохраняется пользователем, сначала этот файл делится на блоки, а затем эти блоки данных распределяются между разными машинами, присутствующими в кластере HDFS. После этого копия каждого блока создается на других машинах, присутствующих в кластере. По умолчанию HDFS создает 3 копии файла на других компьютерах, присутствующих в кластере [8]. Поэтому, если по какой-то причине какое-либо хранилище из кластера HDFS не работает, пользователь может легко получить доступ к сохраненным данным с других компьютеров в кластере, в которых присутствует реплика файла. Таким образом, если при отправке данных возникнет сбой, то это приведет к частично выполненной отправке данных, сами данные, при этом, останутся невредимыми. Однако, факт отсутствия механизма, ликвидирующего этот недостаток при передаче, делает Sqoop плохим кандидатом в плане надежности.

В конечном итоге получаем таблицу систем с их надежностью при передаче данных.

Итоговая таблица надежности систем пакетной обработки

Надежность систем потоковой обработки

Системы потоковой обработки находятся на пике популярности у Big Data инженеров, в основном потому, что являются следующей итерацией систем загрузки данных, оставляя системы пакетной обработки далеко позади. Среди них Apache Flume, Apache Storm, потоковый режим Apache Flink и Apache Apex, Apache NiFi, Fluentd и Apache Kafka.

Начиная с Apache Flume, обратим внимание на то, что события организуются в канале для каждого агента. Затем события передаются в следующий агент или хранилище (например, HDFS) в потоке. События удаляются из канала только после того, как они сохранены в канале следующего агента или в хранилище. Подобный метод доставки сообщений Flume обеспечивает сквозную надежность потока данных. Так же Flume использует транзакционный подход для обеспечения надежной доставки событий. Истоки и стоки используют транзакции для хранения или извлечения событий. Это гарантирует, что множество событий надежно передается от точки к точке в потоке [9].

Архитектура Apache Storm включает в себя мастер-процесс (Nimbus), его супервизор, и рабочие процессы, располагаемые на узлах. Nimbus контролирует рабочие процессы системы. Отказ рабочего, заставит супервизор перезапустить его. Если он постоянно терпит неудачу при запуске и не может связаться с Nimbus, Nimbus переназначит рабочего на другой узел. Если откажет узел — задачи, назначенные этому узлу, истекут по тайм-ауту и Nimbus переназначит эти задачи другим узлам. Если откажет Nimbus рабочие все равно будут продолжать функционировать. Кроме того, супервизор будет продолжать перезапускать рабочих, при их отказе. Однако, без Nimbus, рабочие не будут переназначаться на другие машины, когда это необходимо (например, при отказе узла). Таким образом в архитектуре Apache Storm — Nimbus является точкой отказа системы [10].

Центральная часть механизма отказоустойчивости Flink представляет собой согласованные снимки потока данных и состояний операторов. Эти моментальные снимки действуют как согласованные контрольные точки, на которые система может вернуться в случае сбоя. Механизм Flink для создания этих снимков основан на алгоритме Chandy-Lamport для распределенных снимков и адаптирован к модели исполнения Flink. Восстановление по этому механизму происходит следующим образом: После сбоя Flink выбирает последнюю завершенную контрольную точку k. Затем система повторно развертывает весь распределенный поток данных и предоставляет каждому оператору состояние, которое было сохранено как часть контрольной точки k. Источники настроены на начало чтения потока из положения S^k.Например, в Apache Kafka это означает, что потребитель должен начать получать при смещении S^k [11].

В следующей системе — Apache Apex, мастер потоковой передачи предоставляет параметры контрольной точки StreamingContainer во время инициализации. Период контрольной точки предоставляется контейнерам, которые имеют оконные генераторы. Контрольный кортеж отправляется в конце интервала контрольной точки. Этот кортеж проходит через путь данных через потоки и запускает каждый StreamingContainer на пути к контрольной точке оператора, который получает этот кортеж. Это гарантирует, что все контрольные точки операторов находятся на одной и той же границе окна (за исключением тех случаев, когда пользователь настраивал другой интервал контрольной точки для оператора). Процесс использования контрольных точек включает в себя приостановку оператора, сохранение состояния в постоянном хранилище и возобновление работы оператора. Таким образом, использование контрольных точек создает затраты на задержку, которые могут отрицательно влиять на пропускную способность системы. Чтобы минимизировать это воздействие, важно обеспечить сохранение контрольной точки с минимально необходимыми объектами. Это означает, что, как упоминалось ранее, все данные, которые не являются частью состояния оператора, должны быть объявлены переходными, чтобы они не сохранялись [12].

В NiFi cодержимое потоковых файлов записывается в хранилище контента. Когда узел сбоит, диспетчер кластера NiFi направляет данные на другой узел. Следует отметить,что NiFi не копирует данные, подобно Kafka. Данные ждущие очереди в отказавший узел все равно останутся там. Подобная проблема решается отправлением данных на работающий узел в кластере вручную или же исправлением сбоившего узла. Любые новые данные автоматически перенаправляются на другие узлы кластера NiFi с помощью Cluster Manager (NCM). Очередь данных обычно очень мала, и составляет около 1 секунды полученных данных или даже меньше. Данные в очередях могут быть потеряны, если откажет диск. Как правило, при использовании системы в производственном цикле, настоятельно рекомендуется использовать RAID массивы, которые обеспечивают избыточность. Пока один диск в каждой паре функционирует, данные могут быть восстановлены [13].

Иногда в Fluentd при использовании библиотек журналов на разных языках при сбое приложения журнальные записи не могут быть сохранены в локальном экземпляре Fluentd. В зависимости от зрелости каждой библиотеки журналов были реализованы некоторые механизмы для предотвращения потери данных:

– Экспоненциальный откат (доступен для Ruby и Java). При попытке повторно отправить журналы в локальный Fluentd, его реализации будут использовать экспоненциальный откат для предотвращения чрезмерных запросов повторного подключения.

Если же отказывает процесс Fluentd, тогда в зависимости от конфигурации буфера происходит следующее:

– Если вы используете buf_memory, буферизованные данные полностью потеряны. Это является компромиссом для повышения производительности, поскольку уменьшение flush_interval уменьшит вероятность потери данных, но увеличит количество передач между модулями пересылки и накопления.

– Если вы используете buf_file, буферизованные данные сохраняются на диске. После восстановления Fluentd попытается снова отправить данные в буфере в пункт назначения.

Если хранилище куда направляются данные (например, Amazon S3, MongoDB, HDFS и т. д.) не работает, Fluentd будет продолжать повторять отправку буферизованных данных. Логика повторения зависит от реализации плагина.

При использовании buf_memory, модули накопления перестанут принимать новые журналы, как только они достигнут пределов своего буфера. Однако, при использовании buf_file, модули накопления будут продолжать принимать журналы до тех пор, пока они не исчерпают дисковое пространство [14].

В Kafka каждый раздел реплицируется через настраиваемое количество серверов для отказоустойчивости. Каждый раздел имеет один сервер, который действует как мастер и ноль или более серверов, которые действуют как рабочие. Мастер обрабатывает все запросы на чтение и запись для раздела, а рабочие пассивно реплицируют мастера. Если мастер отказывает, то один из рабочих автоматически становится новым мастером. Каждый сервер выступает в качестве мастера для некоторых своих разделов и для других сторонников, поэтому загрузка внутри кластера хорошо сбалансирована. Так же Kafka гарантирует, что для топика с коэффициентом репликации N, можно защититься от N-1 сбоев не теряя записей, зафиксированных в журнале.

Финальная таблица систем выглядит следующим образом:

Таблица 2

Методы надежности систем загрузки журнальных данных

Следующей особенностью надежности систем потоковой обработки является метод гарантирования обработки сообщений. Это обусловлено их распределённой архитектурой и необходимостью синхронизации данных на разных узлах кластера, что напрямую влияет на производительность. Различают обработку «максимум один раз» (at most once), «по крайней мере один раз» (at least once) и «строго один раз» (exactly once)».Максимум один раз» означает, что сообщение может быть доставлено, а может быть потеряно. «По крайней мере один раз» предполагает, что сообщение будет доставлено на обработку в любом случае, но может оказаться, что такое сообщение уже существует. Очевидный недостаток этого метода — необходимость учитывать возможность появления дубликатов что, соответственно, накладывает дополнительную нагрузку на программиста по их учёту в своих алгоритмах. «Точно один раз» означает, что сообщение будет гарантированно доставлено, и при этом будут исключены дубликаты. Этот метод самый удобный для пользователя, но и, очевидно, самый сложный и медленный в реализации системы потоковой обработки.

Apache Flume имеет более слабые гарантии, чем некоторые другие системы (например, как будет рассмотрено ниже — очереди сообщений) в интересах более быстрого перемещения данных и обеспечения более дешевой отказоустойчивости (основная идея заключается в том, чтобы свести к минимуму количество состояний, которое должен хранить Flume. Реплицированное состояний — это то, что делает отказоустойчивость сложной, и затрудняет определение причин неудачных передач). В режиме надежности Flume события доставляются «по крайней мере один раз», но это сказывается на пропускной способности системы [9].

Apache Storm же предлагает несколько различных уровней гарантированной обработки сообщений, включая «максимум один раз», «по крайней мере один раз», и «строго один раз» с помощью дополнения Trident [14].

Apache Flink предлагает механизм отказоустойчивости для постоянного восстановления состояния приложений для потоковой передачи данных. Механизм гарантирует, что даже при наличии сбоев состояние программы в конечном итоге передаст каждую запись из потока данных «строго один раз» [11].

Apache Apex имеет в своем арсенале несколько уровней гарантированной обработки сообщений, включая «максимум один раз», «по крайней мере один раз», и «строго один раз». Каждый из уровней подразумевает определенную пропускную способность, поэтому прежде, чем планировать поток данных рекомендуется ознакомиться с технической документацией этой системы [12].

C Apache NiFi все не так просто. Прежде всего, говоря о связи между несколькими экземплярами NIFI через протокол «site-to-site», из-за 2-х фазовой передачи, предоставляемой на уровне фреймворка NiFi, будет гарантирована доставка сообщения «как минимум раз». В остальных случаях соединения через S2S вероятность доставки «строго один раз» максимальна. Не гарантируется, только если ваша система разрывается при двух фазовой передаче (буквально между двумя строками кода). Например, после того, как данные доставляются на принимающую сторону, но до того, как отправляющая сторона заявит об этом, сетевое соединение каким-то образом оборвется. В этом случае отправляющая сторона попытается отправить данные снова, когда сетевое подключение будет восстановлено, и на принимающей стороне образуются дубликаты. Одним из способов справиться с этим использовать процессор DetectDuplicate, поместив его сразу после порта S2S. Аналогичным образом, когда протокол S2S NIFI встроен в другие системы, такие как Storm или Spark, гарантируется доставка сообщения «как минимум раз».

Гарантия доставки сообщений на уровне NIFI-обработчика, определяется в зависимости от того, как транспортировка данных обрабатывается этим обработчиком. Например, для внешних систем, поддерживающих двухфазную передачу как в аналогичной S2S — Kafka, гарантируется доставка «как минимум раз». Если другая система не поддерживает двухфазную передачу, например, как протокол Syslog, гарантируется доставка «максимум один раз» [15].

Fluentd гарантирует доставку «максимум один раз» и «по крайней мере один раз». Доставка «строго один раз» не реализована разработчиками, чтобы собирать огромные объемы данных без влияния на производительность приложений. Регистратор данных передает данные асинхронно, что улучшает производительность за счет потенциальных сбоев доставки [16].

Kafka гарантирует доставку вида по «крайней мере один раз» по умолчанию и позволяет пользователю реализовать вид доставки «максимум один раз», отключив повторные попытки отправителя и организовав смещение перед обработкой пакета сообщений. Доставка «строго один раз» зависит от настроек целевой системы хранения [17].

В заключение этого раздела, приведем таблицу гарантий доставки сообщения систем потоковой обработки данных.

Гарантии доставки сообщения систем потоковой обработки данных

Заключение

Хотелось бы отметить недочеты в реализации надежности некоторых систем, тем самым определив явных фаворитов.

Среди систем пакетной обработки явно выделяется Apache Sqoop с отсутствием возможности повторять отправку данных или восстанавливать передачу через контрольную точку. В остальном все системы пакетной обработки имеют, определенный их архитектурой, метод препятствования сбоям при передаче, каждая из них гарантирует сохранность данных.

Среди систем потоковой обработки проигрывает Apache Storm, поскольку, как сказано в технической документации, он обладает точкой отказа системы в виде мастера-Nimbus отказ которого может привести к частичной потере данных при передаче. Остальные системы обладают тем или иным способом защиты данных от сбоев. Выбор определенной системы зависит от нужд Big Data инженера и проекта.

Литература:

1. Performance and reliability in Big Data systems [Электронный ресурс]Режим доступа: https://msdn.microsoft.com/en-us/library/dn749838.aspx#sec3
2. Apache software foundation projects list [Электронный ресурс]Режим доступа: https://www.apache.org/index.html#projects-list
3. Fluentd [Электронный ресурс]Режим доступа: https://www.fluentd.org/
4. Datatorrent [Электронный ресурс]Режим доступа: https://www.datatorrent.com/
5. Apache Chukwa basic collector operation [Электронный ресурс]Режим доступа: http://chukwa.apache.org/docs/r0.5.0/collector.html
6. Apache Apex fault-tolerance [Электронный ресурс]Режим доступа: http://apex.apache.org/docs/malhar/operators/file_output/#fault-tolerance
7. Apache Flink batch processing fault tolerance [Электронный ресурс]Режим доступа: https://ci.apache.org/projects/flink/flink-docs-release-1.4/dev/batch/fault_tolerance.html
8. Hadoop distributed file system failt tolerance [Электронный ресурс]Режим доступа: https://data-flair.training/blogs/learn-hadoop-hdfs-fault-tolerance/
9. Apache Flume user guide [Электронный ресурс]Режим доступа: https://flume.apache.org/FlumeUserGuide.html
10. Apache Storm fault tolerance [Электронный ресурс]Режим доступа: http://storm.apache.org/releases/1.2.1/Daemon-Fault-Tolerance.html
11. Apache Flink stream checkpointing [Электронный ресурс]Режим доступа: https://ci.apache.org/projects/flink/flink-docs-release-1.4/internals/stream_checkpointing.html
12. Apache Apex fault tolerance [Электронный ресурс]Режим доступа: http://apex.apache.org/docs/apex/application_development/#fault-tolerance
13. Apache Nifi fault tolerance [Электронный ресурс]Режим доступа: https://community.hortonworks.com/questions/66385/details-on-nifi-fault-tolerance-details.html
14. Fluentd failure scenarios [Электронный ресурс]Режим доступа: https://docs.fluentd.org/v0.12/articles/failure-scenarios
15. Apache Storm guaranteed message processing [Электронный ресурс]Режим доступа: http://storm.apache.org/releases/current/Guaranteeing-message-processing.html
16. Apache NiFi message delivery guarantees [Электронный ресурс]Режим доступа: https://community.hortonworks.com/articles/87948/at-least-once-delivery-vs-exactly-once-delivery-se.html
17. Fluentd message delivery semantics [Электронный ресурс]Режим доступа: https://docs.fluentd.org/v0.12/articles/high-availability#message-delivery-semantics
18. Apache Kafka semantics [Электронный ресурс]Режим доступа: https://kafka.apache.org/08/documentation.html#semantics